Examples

Acme Tickets is a connector for a made-up ticket system with a JSON API. It has three tools: list tickets, read one ticket’s text, and open a ticket. Save it as connectors/acme.py or connectors/acme.star in your library.

Python

connectors/acme.py

"""Acme Tickets: our ticket system, as a Flowplane connector."""

from __future__ import annotations

from typing import Annotated, Literal

from console_api.connectors.core import (
    ALL_MODES,
    ApiKey,
    BaseConnector,
    Category,
    Check,
    Errors,
    address,
    credential,
    tool,
)
from console_api.connectors.core.script import fail, ok, request
from console_api.connectors.lib.shape import dig

_ID = "The ticket's id, from list_tickets."


class Acme(BaseConnector):
    id = "custom_acme"
    name = "Acme Tickets"
    summary = "Look up and open tickets in Acme."
    category = Category.PRODUCTIVITY
    icon = "acme"
    docs = "https://acme.example.com/docs/api"
    fields = (
        address(
            label="Acme address",
            placeholder="https://tickets.acme.example.com",
            help="Where you open Acme.",
        ),
    )
    auth = (ApiKey(credential("API key", min_length=16, help="Settings, API keys.")),)
    modes = ALL_MODES
    check = Check("/api/me")
    errors = Errors(said=("error",), hints={404: "Nothing there, or the key cannot see it."})

    @tool.script(intent="read")
    def list_tickets(
        status: Annotated[Literal["open", "closed"], "Only tickets in this state."] = None,
        query: Annotated[str, "Words in the title."] = None,
    ):
        """The tickets the key can see, newest first."""
        params = {}
        if status:
            params["status"] = status
        if query:
            params["q"] = query
        tickets = ok(request("GET", "/api/tickets", query=params))["data"]
        return [{"id": t["id"], "title": t["title"], "status": t["status"]} for t in tickets]

    @tool.script(intent="read")
    def ticket_text(ticket_id: Annotated[str, _ID]):
        """The readable text of a ticket and its replies."""
        answer = request("GET", "/api/tickets/" + ticket_id)
        if answer.status == 404:
            fail("no such ticket")
        ticket = ok(answer)
        return {"text": dig(ticket, "body"), "replies": [r["text"] for r in ticket["replies"]]}

    @tool.script(intent="write", ask=True)
    def open_ticket(
        title: Annotated[str, "A short title."],
        body: Annotated[str, "What happened."],
    ):
        """Open a ticket."""
        ticket = ok(request("POST", "/api/tickets", body={"title": title, "body": body}))
        return {"id": ticket["id"]}

Starlark

connectors/acme.star

# Acme Tickets: our ticket system, as a Flowplane connector.
load("//lib/shape", "dig")

def list_tickets(status = None, query = None):
    params = {}
    if status:
        params["status"] = status
    if query:
        params["q"] = query
    tickets = ok(request("GET", "/api/tickets", query = params))["data"]
    return [{"id": t["id"], "title": t["title"], "status": t["status"]} for t in tickets]

def ticket_text(ticket_id):
    answer = request("GET", "/api/tickets/" + ticket_id)
    if answer.status == 404:
        fail("no such ticket")
    ticket = ok(answer)
    return {"text": dig(ticket, "body"), "replies": [r["text"] for r in ticket["replies"]]}

def open_ticket(title, body):
    ticket = ok(request("POST", "/api/tickets", body = {"title": title, "body": body}))
    return {"id": ticket["id"]}

connector(
    id = "custom_acme",
    name = "Acme Tickets",
    summary = "Look up and open tickets in Acme.",
    category = "productivity",
    fields = [address_field(label = "Acme address", placeholder = "https://tickets.acme.example.com")],
    auth = [api_key("API key", min_length = 16)],
    check = check("/api/me"),
    errors = errors(said = ["error"], hints = {404: "Nothing there, or the key cannot see it."}),
    tools = [
        script_tool(
            "list_tickets", "The tickets the key can see, newest first.", list_tickets,
            intent = "read",
            params = [
                param("status", description = "Only tickets in this state.", enum = ["open", "closed"]),
                param("query", description = "Words in the title."),
            ],
        ),
        script_tool(
            "ticket_text", "The readable text of a ticket and its replies.", ticket_text,
            intent = "read",
            params = [param("ticket_id", required = True, description = "The ticket's id, from list_tickets.")],
        ),
        script_tool(
            "open_ticket", "Open a ticket.", open_ticket,
            intent = "write", ask = True,
            params = [
                param("title", required = True, description = "A short title."),
                param("body", required = True, description = "What happened."),
            ],
        ),
    ],
)

What happens next

1. Push the file to the branch or tag your library follows.
2. In the console, the library lists Acme Tickets, or the line and reason of any error.
3. Install it on a tunnel that can reach tickets.acme.example.com.

The Starlark format is deliberately small: one API-key login, no paging or sign-in steps. Use Python when you need those.