Your First CRAFT Agent

This tutorial builds the same agent three times — once for each capability layer — so you can stop at whichever level you need. By the end you have a registered CRAFT agent that lists your project’s data connections and is discoverable by other platform services. What you’ll build: A data-assistant agent that:

Responds to prompts using CRAFT’s managed LLM gateway
Calls the CRAFT Assets API to list real data connections
Is registered in the CRAFT agent registry

Time: ~5 min per step. Steps are independent — come back later for the next one. Prerequisites:

Obtain a CRAFT project access token via your deployment’s OIDC token endpoint (client credentials grant) and export the connection settings:

export CRAFT_TOKEN="$(curl -s -X POST "${OIDC_TOKEN_URL}" \
  -d "grant_type=client_credentials&client_id=${OIDC_CLIENT_ID}&client_secret=${OIDC_CLIENT_SECRET}" \
  | python3 -c 'import sys,json; print(json.load(sys.stdin)["access_token"])')"
export CRAFT_GATEWAY_URL="<your-llm-gateway-url>"    # LiteLLM sidecar or gateway
export CRAFT_ASSETS_URL="<your-assets-api-url>"      # CRAFT Assets API
export CRAFT_PROJECT_ID="<your-project-uuid>"

For local dev, run docker-compose up from the solution starter template — it pre-configures OIDC_TOKEN_URL, OIDC_CLIENT_ID, OIDC_CLIENT_SECRET, and the local URLs.

Step 1 — Connect to CRAFT’s LLM gateway

The only difference between a standalone agent and a CRAFT-native agent is where it routes its LLM calls. CRAFT runs a LiteLLM gateway that handles model selection, project billing, and rate limits. All four frameworks reach it through an OpenAI-compatible endpoint. Pick your framework:

Google ADK
Claude Agent SDK
Pydantic AI
LangGraph

pip install google-adk litellm

agent.py

import os
from google.adk.agents import Agent
from google.adk.models.lite_llm import LiteLlm
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

# Point ADK at CRAFT's LiteLLM gateway via the LiteLlm wrapper.
# The "openai/" prefix tells LiteLLM to use the OpenAI-compatible endpoint.
model = LiteLlm(
    model="openai/claude-opus-4-8",
    api_base=os.environ["CRAFT_GATEWAY_URL"],
    api_key=os.environ["CRAFT_TOKEN"],
)

agent = Agent(
    name="data_assistant",
    model=model,
    description="Helps users understand the data available in their CRAFT project.",
    instruction="You are a helpful data assistant. Answer questions about the user's data concisely.",
)

async def main():
    session_service = InMemorySessionService()
    session = await session_service.create_session(
        app_name="data_assistant", user_id="dev", session_id="s1"
    )
    runner = Runner(agent=agent, app_name="data_assistant", session_service=session_service)

    async for event in runner.run_async(
        user_id="dev", session_id="s1",
        new_message=types.Content(role="user", parts=[types.Part(text="What can you help me with?")]),
    ):
        if event.is_final_response():
            print(event.content.parts[0].text)

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

For production multi-agent patterns with A2A, see the multi-agent patterns guide — it shows how an orchestrator uses ADK to chain a text-to-SQL agent and an insights agent over the A2A protocol.

pip install anthropic openai

agent.py

import os
from openai import OpenAI

# Route through CRAFT's LiteLLM gateway using the OpenAI-compatible client.
# LiteLLM serves claude-opus-4-8 (and other models) at the standard
# /v1/chat/completions endpoint — no Anthropic account needed.
client = OpenAI(
    base_url=os.environ["CRAFT_GATEWAY_URL"],
    api_key=os.environ["CRAFT_TOKEN"],
)

def run(prompt: str) -> str:
    response = client.chat.completions.create(
        model="claude-opus-4-8",
        max_tokens=1024,
        messages=[
            {"role": "system", "content": "You are a helpful data assistant. Answer questions about the user's data concisely."},
            {"role": "user", "content": prompt},
        ],
    )
    return response.choices[0].message.content

if __name__ == "__main__":
    print(run("What can you help me with?"))

The OpenAI client talks to CRAFT’s LiteLLM gateway using Claude via the OpenAI-compatible endpoint. LiteLLM translates function-calling requests to the format each model expects, so the same pattern works for any model in the CRAFT allowlist.

pip install pydantic-ai

agent.py

import os
from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.providers.openai import OpenAIProvider

# Route through CRAFT's LiteLLM gateway using the OpenAI-compatible provider.
model = OpenAIChatModel(
    "claude-opus-4-8",
    provider=OpenAIProvider(
        base_url=os.environ["CRAFT_GATEWAY_URL"],
        api_key=os.environ["CRAFT_TOKEN"],
    ),
)

agent = Agent(
    model,
    system_prompt="You are a helpful data assistant. Answer questions about the user's data concisely.",
)

result = agent.run_sync("What can you help me with?")
print(result.output)

Pydantic AI’s FastMCPToolset wires MCP servers as tools with first-class typing. See the tool authoring guide for that pattern.

pip install langgraph langchain-openai

agent.py

import os
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
from langchain_core.messages import HumanMessage

# Route through CRAFT's LiteLLM gateway — it's OpenAI-compatible.
llm = ChatOpenAI(
    model="claude-opus-4-8",
    base_url=os.environ["CRAFT_GATEWAY_URL"],
    api_key=os.environ["CRAFT_TOKEN"],
)

agent = create_react_agent(
    model=llm,
    tools=[],  # add tools in Step 2
    prompt="You are a helpful data assistant. Answer questions about the user's data concisely.",
)

result = agent.invoke({"messages": [HumanMessage(content="What can you help me with?")]})
print(result["messages"][-1].content)

That’s it. Your agent is running on CRAFT’s managed LLM infrastructure.

Routing to specific backends

The CRAFT gateway uses LiteLLM under the hood, so any model string it recognises works in the model= field — you don’t need to change any client code. Ask your platform team which models are in the project’s allowlist.

Vertex AI (Google Cloud) — CRAFT default

# Model string format: vertex_ai/<model-id>
# LiteLLM reads VERTEXAI_PROJECT + VERTEXAI_LOCATION (or the Google Cloud
# standards GOOGLE_CLOUD_PROJECT + GOOGLE_CLOUD_LOCATION) and authenticates
# via Workload Identity or GOOGLE_APPLICATION_CREDENTIALS automatically.

# DEFAULT — fastest, highest quality on agentic + coding benchmarks
model = "vertex_ai/gemini-3.5-flash"

# SPECIALIZED — only if your agent depends on extensive custom tool calling
# (the model prioritizes your registered functions over bash fallbacks)
model = "vertex_ai/gemini-3.1-pro-preview-customtools"

# SECONDARY — demonstrates model agnosticity
model = "vertex_ai/claude-opus-4-8"

Use gemini-3.1-pro-preview-customtools only when extensive custom tool calling is core to your agent. Same intelligence as base gemini-3.1-pro, fine-tuned to prioritize registered custom functions over bash fallbacks. Google’s own guidance: if >50% of requests don’t involve tool calling, stay on gemini-3.5-flash — 3.5-flash already beats 3.1-pro on agentic benchmarks (MCP Atlas +5.4%) and coding (Terminal-Bench +6%) at 3.6× the speed (source), and the customtools variant degrades quality on non-tool workloads (source). When you do use it: preview SLA, global endpoint only (vertex_location must be global), lower quota than GA — configure a fallback to gemini-3.5-flash or claude-opus-4-8.

Amazon Bedrock

# Model string format: bedrock/<model-id>
# LiteLLM uses AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY or instance profile.
model = "bedrock/anthropic.claude-opus-4-8-20260301-v1:0"

Azure AI Foundry

# Model string format: azure/<deployment-name>
# LiteLLM reads AZURE_API_KEY and AZURE_API_BASE from the gateway config.
model = "azure/gpt-5.5"

Nebius Tokenfactory

# Nebius exposes an OpenAI-compatible endpoint; prefix with openai/.
# LiteLLM reads NEBIUS_API_KEY from the gateway config.
model = "openai/Qwen/Qwen3-30B-A3B"

Step 2 — Add a CRAFT tool

Tools let your agent take action. The simplest CRAFT tool calls the Assets API to list the data connections registered in your project — real databases and warehouses the platform knows about.

Google ADK
Claude Agent SDK
Pydantic AI
LangGraph

agent.py

import os
import httpx
from google.adk.agents import Agent
from google.adk.models.lite_llm import LiteLlm
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

# ── CRAFT tool ─────────────────────────────────────────────────────────
def list_data_connections() -> str:
    """List the data connections available in the current CRAFT project."""
    resp = httpx.get(
        f"{os.environ['CRAFT_ASSETS_URL']}/assets/data",
        headers={
            "Authorization": f"Bearer {os.environ['CRAFT_TOKEN']}",
            "X-Project-ID": os.environ["CRAFT_PROJECT_ID"],
        },
    )
    resp.raise_for_status()
    items = resp.json().get("data", [])
    if not items:
        return "No data connections found in this project."
    return "\n".join(f"- {c['name']} ({c['connection_type']})" for c in items)

# ── Agent ──────────────────────────────────────────────────────────────
model = LiteLlm(
    model="openai/claude-opus-4-8",
    api_base=os.environ["CRAFT_GATEWAY_URL"],
    api_key=os.environ["CRAFT_TOKEN"],
)

agent = Agent(
    name="data_assistant",
    model=model,
    description="Helps users understand the data available in their CRAFT project.",
    instruction="You are a helpful data assistant. Use list_data_connections to answer questions about available data.",
    tools=[list_data_connections],
)

async def main():
    session_service = InMemorySessionService()
    session = await session_service.create_session(
        app_name="data_assistant", user_id="dev", session_id="s1"
    )
    runner = Runner(agent=agent, app_name="data_assistant", session_service=session_service)

    async for event in runner.run_async(
        user_id="dev", session_id="s1",
        new_message=types.Content(role="user", parts=[types.Part(text="What data connections do I have?")]),
    ):
        if event.is_final_response():
            print(event.content.parts[0].text)

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

agent.py

import os
import json
import httpx
from openai import OpenAI

# Same CRAFT gateway client as Step 1 — tool calling works over the
# OpenAI-compatible endpoint via LiteLLM's function-calling support.
client = OpenAI(
    base_url=os.environ["CRAFT_GATEWAY_URL"],
    api_key=os.environ["CRAFT_TOKEN"],
)

# ── CRAFT tool definition (OpenAI function-calling format) ──────────────
TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "list_data_connections",
            "description": (
                "List all data connections (databases, data warehouses, APIs) registered "
                "in the current CRAFT project. Call this whenever the user asks what data "
                "is available."
            ),
            "parameters": {"type": "object", "properties": {}, "required": []},
        },
    }
]

def list_data_connections() -> str:
    resp = httpx.get(
        f"{os.environ['CRAFT_ASSETS_URL']}/assets/data",
        headers={
            "Authorization": f"Bearer {os.environ['CRAFT_TOKEN']}",
            "X-Project-ID": os.environ["CRAFT_PROJECT_ID"],
        },
    )
    resp.raise_for_status()
    items = resp.json().get("data", [])
    if not items:
        return "No data connections found in this project."
    return "\n".join(f"- {c['name']} ({c['connection_type']})" for c in items)

TOOL_HANDLERS = {"list_data_connections": list_data_connections}

# ── Agentic loop ────────────────────────────────────────────────────────
def run(prompt: str) -> str:
    messages = [
        {"role": "system", "content": "You are a helpful data assistant. Use your tools to answer questions about available data."},
        {"role": "user", "content": prompt},
    ]
    while True:
        response = client.chat.completions.create(
            model="claude-opus-4-8",
            max_tokens=1024,
            tools=TOOLS,
            messages=messages,
        )
        msg = response.choices[0].message
        messages.append(msg)

        if not msg.tool_calls:
            return msg.content

        # Execute tool calls and feed results back
        results = []
        for tc in msg.tool_calls:
            output = TOOL_HANDLERS[tc.function.name](**json.loads(tc.function.arguments or "{}"))
            results.append({
                "role": "tool",
                "tool_call_id": tc.id,
                "content": output,
            })
        messages.extend(results)

if __name__ == "__main__":
    print(run("What data connections do I have?"))

agent.py

import os
import httpx
from pydantic_ai import Agent
from pydantic_ai.models.openai import OpenAIChatModel
from pydantic_ai.providers.openai import OpenAIProvider

model = OpenAIChatModel(
    "claude-opus-4-8",
    provider=OpenAIProvider(
        base_url=os.environ["CRAFT_GATEWAY_URL"],
        api_key=os.environ["CRAFT_TOKEN"],
    ),
)

agent = Agent(
    model,
    system_prompt="You are a helpful data assistant. Use your tools to answer questions about available data.",
)

# ── CRAFT tool ─────────────────────────────────────────────────────────
@agent.tool_plain
def list_data_connections() -> str:
    """List all data connections registered in the current CRAFT project."""
    resp = httpx.get(
        f"{os.environ['CRAFT_ASSETS_URL']}/assets/data",
        headers={
            "Authorization": f"Bearer {os.environ['CRAFT_TOKEN']}",
            "X-Project-ID": os.environ["CRAFT_PROJECT_ID"],
        },
    )
    resp.raise_for_status()
    items = resp.json().get("data", [])
    if not items:
        return "No data connections found in this project."
    return "\n".join(f"- {c['name']} ({c['connection_type']})" for c in items)

result = agent.run_sync("What data connections do I have?")
print(result.output)

@agent.tool_plain registers the function as a tool with no dependency injection. Pydantic AI infers the JSON schema from the docstring and type hints automatically.

agent.py

import os
import httpx
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage
from langgraph.prebuilt import create_react_agent

llm = ChatOpenAI(
    model="claude-opus-4-8",
    base_url=os.environ["CRAFT_GATEWAY_URL"],
    api_key=os.environ["CRAFT_TOKEN"],
)

# ── CRAFT tool ─────────────────────────────────────────────────────────
@tool
def list_data_connections() -> str:
    """List all data connections registered in the current CRAFT project.
    Call this whenever the user asks what data sources or databases are available.
    """
    resp = httpx.get(
        f"{os.environ['CRAFT_ASSETS_URL']}/assets/data",
        headers={
            "Authorization": f"Bearer {os.environ['CRAFT_TOKEN']}",
            "X-Project-ID": os.environ["CRAFT_PROJECT_ID"],
        },
    )
    resp.raise_for_status()
    items = resp.json().get("data", [])
    if not items:
        return "No data connections found in this project."
    return "\n".join(f"- {c['name']} ({c['connection_type']})" for c in items)

agent = create_react_agent(
    model=llm,
    tools=[list_data_connections],
    prompt="You are a helpful data assistant. Use your tools to answer questions about available data.",
)

result = agent.invoke({"messages": [HumanMessage(content="What data connections do I have?")]})
print(result["messages"][-1].content)

Your agent now answers questions using real platform data. Every other CRAFT capability — agents, files, models, artifacts — is one more httpx.get() away.

Step 3 — Register your agent

A registered agent is discoverable by other platform services, the UI, and other agents via the A2A protocol. Registration is one POST to the Assets API.

import os
import httpx

# POST /assets/agents expects the full Agent Card wrapped in "agent_card".
# "name" and "version" are required; "url" points to your running service's
# /.well-known/agent-card.json endpoint so other agents can fetch full details.
registration = {
    "agent_card": {
        "name": "data-assistant",
        "version": "1.0.0",
        "description": "Answers questions about data connections in a CRAFT project.",
        "url": "http://your-service-host/.well-known/agent-card.json",
        "capabilities": {
            "streaming": False,
            "push_notifications": False,
        },
    },
    "tags": ["data", "assistant"],
}

resp = httpx.post(
    f"{os.environ['CRAFT_ASSETS_URL']}/assets/agents",
    json=registration,
    headers={
        "Authorization": f"Bearer {os.environ['CRAFT_TOKEN']}",
        "X-Project-ID": os.environ["CRAFT_PROJECT_ID"],
    },
)
resp.raise_for_status()
agent_id = resp.json()["resource_uri"]
print(f"Registered: {agent_id}")

The agent_card_url points to a /.well-known/agent-card.json endpoint your service exposes — it describes your agent’s skills so other agents can call it. See the A2A protocol primer for the Agent Card schema, and multi-agent patterns for wiring A2A delegation.

Re-POSTing the same name in a project returns 409 RESOURCE_ALREADY_EXISTS. To update an existing registration, PUT /assets/agents/{resource_uri} and include the current ETag in If-Match:

PUT without If-Match returns 428 Precondition Required.
PUT with a stale If-Match returns 412 Precondition Failed ("ETag mismatch — resource was modified").

Read the current ETag from the ETag header on a prior GET /assets/agents/{resource_uri} (or use the current_version field from the JSON body — both are equivalent), and feed it back on the next PUT. Either the bare integer (If-Match: 1) or the quoted form (If-Match: "1") is accepted; weak ETags (W/"1") are rejected. Automate the GET → PUT(If-Match) cycle in your deploy pipeline so the registry stays in sync with your running service.

What’s next

Add more tools

Function tools, MCP tools via FastMCPToolset, schema discipline.

Multi-agent patterns

A2A delegation, parallel fan-out, supervisor agents.

A2A protocol primer

Agent Cards, JSON-RPC over SSE, task lifecycle.

Eval harness

Golden traces, Langfuse evaluators, regression suites.

​Step 1 — Connect to CRAFT’s LLM gateway

​Routing to specific backends

​Step 2 — Add a CRAFT tool

​Step 3 — Register your agent

​What’s next

Add more tools

Multi-agent patterns

A2A protocol primer

Eval harness

Step 1 — Connect to CRAFT’s LLM gateway

Routing to specific backends

Step 2 — Add a CRAFT tool

Step 3 — Register your agent

What’s next