---
name: agenthub-python
description: Build and run Agent Hub workflows with the agentsapi Python client—execute agents (polling vs SSE), local tools and webhooks, output validators (StandardAgentOutput, upsert_validators), and persist runs locally. Use when writing Python against Agent Hub, agentserviceapi, agentsapi, AgentServiceAPIClient, tool webhooks, or workflow execution.
---

# Agent Hub — Python client (`agentsapi`)

Use the **Agent Service** Python package: **PyPI name `agentserviceapi`**, import **`agentsapi`**.

```bash
pip install agentserviceapi
```

```python
from agentsapi import AgentServiceAPIClient

client = AgentServiceAPIClient(base_url="https://your-agent-hub-host")  # default in library may differ
client.set_api_key("sk_...")  # programmatic key from Agent Hub
```

For **session-cookie** auth (browser flows), use `set_auth_cookie` instead of or in addition to API key, per product docs.

---

## Execute workflows

### Fire-and-forget + poll or webhook

Start a run and get a `task_id` immediately (no SSE loop on the client):

```python
resp = await client.execute_agent_polling(
    agent_id="<workflow-uuid>",
    string_inputs={"1": "user text slot 1"},
    image_inputs={},
    environment="playground",  # or "test" | "production"
)
# resp.task_id — poll with get_agent_task_status or observe via webhooks
```

### Interactive run (SSE + tools on this machine)

Requires Redis/SSE on the API. Register tool handlers, then run:

```python
from agentsapi.models.tools import ToolDefinition, ParametersSchema, ParameterProperty

async def my_tool(args, *, agent_id, task_id):
    # Return a string (JSON string is fine)
    return "done"

client.set_tools({
    "my_tool": (
        ToolDefinition(
            name="my_tool",
            description="...",
            schema_data=ParametersSchema(
                type="object",
                properties={"q": ParameterProperty(type="string", description="...")},
                required=["q"],
            ),
        ),
        my_tool,
    ),
})

final = await client.run_agent(
    agent_id="<workflow-uuid>",
    string_inputs={"1": "hello"},
    environment="playground",
)
# final is AgentGraphRunStatusResponse
```

Tool handler contract matches webhooks: **`async def handler(args, *, agent_id, task_id) -> str`**. Non-string results are JSON-serialized.

Use `execute_agent` if you need the `AgentInteractiveRun` handle; `run_agent` is `execute_agent` then `await` the result.

Optional: `upload_run_input_file` for file slots before executing.

---

## Tool webhooks (your HTTP server)

When the runner pauses for tools and POSTs to your URL, parse the body into the same model used by the API:

```python
from agentsapi.models.agent_execution_updates import AgentGraphRunStatusResponse

payload = AgentGraphRunStatusResponse.model_validate(await request.json())

tool_results = await client.process_webhook_tools(payload, tools_dict)
await client.send_tool_responses(
    payload.agent_id,
    payload.task_id,
    tool_results.results,
)
```

`tools_dict` is the same mapping as `set_tools`: `name -> (ToolDefinition | None, async callable)`.

If the webhook is for a **parent** run that spawned a subgraph, check `payload.subgraph_task_id` to link child runs without polling.

Coordinator-style tools can be configured to **always** hit your webhook (see `coordinator_tool_contracts` in the package). Prefer **`AgentGraphRunStatusResponse`** for new code; avoid legacy `extract_from_webhook` unless you know the old payload shape.

---

## Completion webhooks

```python
from agentsapi.models.agent_execution_updates import AgentTaskCompletedProcessedResult

async def on_done(req: AgentGraphRunStatusResponse) -> AgentTaskCompletedProcessedResult:
    return AgentTaskCompletedProcessedResult(
        status_message="ok",
        data={"your": "payload"},
    )

result = await client.process_webhook_agent_completed(
    payload,
    agent_handlers={"<workflow-uuid>": on_done},
)
```

---

## Output validators

### Standard platform shape

Subclass **`StandardAgentOutput`** from `agentsapi.models.agent_execution` so routing fields (`status`, `work_log`, …) stay aligned with graph edges:

```python
from pydantic import Field
from agentsapi.models.agent_execution import StandardAgentOutput

class MyStepOutput(StandardAgentOutput):
    result: MyResultModel = Field(...)  # tighten or structure `result`
```

Register JSON Schema with the API (by Pydantic class):

```python
await client.agent_validation.upsert_validators([MyStepOutput])
```

Then set the node’s **`output_format_validator_name`** (or id) in the workflow to that model’s name / created validator.

Alternatives: `create_validator`, `list_validators`, `get_validator`.

Optional **local** check of a raw model string: `AgentNodeOutputFormatValidator(json_schema={...}).validate_response_output(text)` — requires the **`jsonschema`** package if you use that helper.

---

## Persist runs locally

There is no built-in on-disk store. Serialize the final status:

```python
import json
from pathlib import Path

path = Path("runs") / f"{final.task_id}.json"
path.write_text(
    json.dumps(final.model_dump(mode="json"), default=str, indent=2),
    encoding="utf-8",
)
```

For polling-started runs, fetch the workflow run JSON from the API and optionally normalize:

```python
from agentsapi.models.agent_execution_updates import agent_graph_run_status_from_workflow_run_api

final = agent_graph_run_status_from_workflow_run_api(raw_dict)
```

---

## First run and dashboard

Executing a workflow at least once unlocks fuller dashboard experience (metrics, monitoring, optional GitHub connector issues). Runs are recorded on the service when you execute.

---

## Tool loop (SSE vs webhook)

```mermaid
sequenceDiagram
  participant App as ClientApp
  participant API as AgentHubAPI
  App->>API: POST execute workflow
  alt Interactive_SSE
    API-->>App: SSE paused_for_tools
    App->>App: process_webhook_tools
    App->>API: POST tool_responses
  else Tool_webhook
    API->>App: POST webhook JSON
    App->>App: AgentGraphRunStatusResponse
    App->>API: POST tool_responses
  end
  API-->>App: terminal status
  App->>App: optional save model_dump JSON
```
