Connections

from shipit_agent import ConnectionRegistry

registry = ConnectionRegistry(credential_store=store, tools=agent.tools)
print(registry.render())

✓ Slack — connected
      tools: slack
  · BigQuery — not connected
      tools: warehouse
      → Connect BigQuery — it needs you to sign in.
  ⧗ Gmail — expired
      → Reconnect Gmail — its token has run out.

agent.run("Post the release note to #eng.")

{"type": "connection_requested",
 "payload": {"connection_id": "warehouse",
             "title": "BigQuery — analytics.usage",
             "reason": "Read the usage tables to total revenue by region.",
             "auth": "oauth",
             "tool": "connections"}}

registry.resolve("warehouse", accepted=True, credential="wh-key-123", by="you")

registry.resolve("warehouse", accepted=False)     # denied, and closed

registry.resolve("slack", accepted=True, credential="xoxb-…")
registry.resolve("slack", accepted=True, credential={"token": "…", "team": "acme"})
registry.resolve("slack", accepted=True, credential=CredentialRecord(...))

@app.post("/connections/{connection_id}")
def answer(connection_id: str, body: AnswerBody):
    registry.resolve(
        connection_id,
        accepted=body.accepted,
        credential=body.credential,
        by=current_user.email,
    )

for request in registry.pending_requests():
    print(request.title, "—", request.reason)

registry.all()            # every connection and its state
registry.connected()      # the usable ones
registry.needing_action() # the ones a human must fix
registry.is_connected("slack")
registry.summary()        # counts, for a header

agent.run("Which of my connections need attention?")

⚯ Checked connections                             2 need action

from shipit_agent import Agent
from shipit_agent.builtins import get_builtin_tool_map
from shipit_agent.integrations import InMemoryCredentialStore

tools = get_builtin_tool_map(llm=llm, project_root=".")

agent = Agent(
    llm=llm,
    tools=[tools["connections"], tools["slack"], tools["google_sheets"]],
    credential_store=InMemoryCredentialStore(),   # or your own store
)

agent.run("What can you reach right now, and what needs setting up?")

⚯ Checked connections                              1 connected · 2 need action

from shipit_agent import ConnectionRegistry

registry = ConnectionRegistry(credential_store=store, tools=agent.tools)
print(registry.render())

for event in agent.stream("Post the Q2 numbers to #finance."):
    if event.type == "connection_requested":
        card = event.payload
        print(f"CONNECT {card['title']}: {card['reason']} ({card['auth']})")

CONNECT Slack: Post the Q2 revenue summary to the #finance channel. (oauth)

from shipit_agent.narrate import render_chat_html
from IPython.display import HTML

HTML(render_chat_html(events, title="Q2 numbers"))

@app.post("/connections/{connection_id}/accept")
def accept(connection_id: str, credential: str, user=Depends(current_user)):
    registry.resolve(
        connection_id, accepted=True, credential=credential, by=user.email
    )
    return {"state": registry.get(connection_id).state.value}   # "connected"

@app.post("/connections/{connection_id}/deny")
def deny(connection_id: str, user=Depends(current_user)):
    registry.resolve(connection_id, accepted=False, by=user.email)

agent.run("Post the Q2 numbers to #finance.")     # now it just works

@app.get("/oauth/slack/callback")
def slack_callback(code: str):
    token = exchange_code_for_token(code)          # your own exchange
    registry.resolve(
        "slack",
        accepted=True,
        credential={"token": token["access_token"], "team": token["team"]["id"]},
        by="oauth",
    )
    return RedirectResponse("/chat")

class Warehouse:
    name = "warehouse"
    description = "Query the analytics warehouse."
    prompt_instructions = ""
    credential_key = "warehouse"          # ← this is the whole requirement

    def schema(self): ...
    def run(self, context, **kwargs): ...

· Warehouse — not connected
      tools: warehouse
      → Connect Warehouse.