SDK Reference

pip install gitagent

from gitagent import Agent

agent = Agent(agent_dir="./my-agent")
response = agent.run("What's the weather in London?")
print(response.output)

from gitagent import Agent

agent = Agent(
    agent_dir="./my-agent",          # path to agent directory
    model_override="gpt-4o-mini",    # override agent.yaml model
    environment="production",         # tag for telemetry
    memory_user_id="user123",        # scope memory to a user
    session_id="session-abc",        # continue an existing session
    metadata={"customer": "acme"},   # arbitrary run metadata
)

response = agent.run(
    input="Your prompt here",
    session_id="optional-session",    # override session from constructor
    metadata={"key": "value"},        # merge with constructor metadata
    timeout=120,                       # seconds (default: from agent.yaml)
)

import asyncio

async def main():
    agent = Agent(agent_dir="./my-agent")
    response = await agent.run_async("What's the capital of France?")
    print(response.output)

asyncio.run(main())

for chunk in agent.stream("Write a poem about the ocean"):
    print(chunk, end="", flush=True)
print()  # newline after streaming

response = agent.run_flow(
    flow_name="research-report",
    input={"topic": "quantum computing"}
)
print(response.output)

response.output          # str: the agent's final response
response.run_id          # str: unique run identifier
response.session_id      # str: session identifier
response.skill_calls     # list[SkillCall]: skills that were called
response.duration_ms     # int: total run duration
response.usage           # TokenUsage: token counts and estimated cost
response.metadata        # dict: metadata attached to this run
response.error           # str | None: error message if run failed
response.success         # bool: True if no error

call = response.skill_calls[0]
call.skill_name      # str
call.input           # dict
call.output          # any
call.duration_ms     # int
call.error           # str | None

usage = response.usage
usage.input_tokens        # int
usage.output_tokens       # int
usage.total_tokens        # int
usage.estimated_cost_usd  # float

from fastapi import FastAPI
from pydantic import BaseModel
from gitagent import Agent

app = FastAPI()
agent = Agent(agent_dir="./my-agent")

class ChatRequest(BaseModel):
    message: str
    session_id: str | None = None
    user_id: str | None = None

@app.post("/chat")
async def chat(request: ChatRequest):
    response = await agent.run_async(
        input=request.message,
        session_id=request.session_id,
        memory_user_id=request.user_id
    )
    return {
        "output": response.output,
        "session_id": response.session_id,
        "run_id": response.run_id
    }

from fastapi.responses import StreamingResponse

@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
    async def generate():
        async for chunk in agent.stream_async(request.message):
            yield f"data: {chunk}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

from gitagent import Agent, skill

@skill(name="my_tool", description="A custom tool")
def my_tool(input: str) -> str:
    return f"Processed: {input}"

agent = Agent(agent_dir="./my-agent")
agent.register_skill(my_tool)

response = agent.run("Use my_tool to process 'hello'")

from gitagent import Agent
from gitagent.hooks import before_skill

@before_skill
def my_hook(context):
    print(f"Calling skill: {context.skill_name}")

agent = Agent(agent_dir="./my-agent")
agent.register_hook(my_hook)

from gitagent import Agent
from gitagent.exceptions import AgentError, SkillError, TimeoutError

try:
    response = agent.run("...", timeout=30)
except TimeoutError:
    print("Agent timed out")
except AgentError as e:
    print(f"Agent error: {e}")

# Or check the response object
if not response.success:
    print(f"Run failed: {response.error}")

import asyncio
from gitagent import Agent

agent = Agent(agent_dir="./my-agent")

prompts = ["Summarize topic A", "Summarize topic B", "Summarize topic C"]

async def run_all():
    tasks = [agent.run_async(p) for p in prompts]
    return await asyncio.gather(*tasks)

responses = asyncio.run(run_all())
for r in responses:
    print(r.output)