Deploy AI Agents
with One Codebase

from kiboup import KiboAgentApp
from langchain_openai import ChatOpenAI
from langgraph.graph import START, MessagesState, StateGraph

app = KiboAgentApp(api_keys={"sk-chat-abc": "chat"})
llm = ChatOpenAI(model="gpt-4o-mini", streaming=True)

builder = StateGraph(MessagesState)
def chatbot(state):
    return {"messages": [llm.invoke(state["messages"])]}
builder.add_node("chatbot", chatbot)
builder.add_edge(START, "chatbot")
graph = builder.compile()

@app.entrypoint
async def invoke(payload, context):
    messages = payload.get("messages",
        [{"role": "user", "content": payload["prompt"]}])

    async def token_stream():
        async for event in graph.astream_events(
            {"messages": messages}, version="v2"
        ):
            if event.get("event") == "on_chat_model_stream":
                content = event["data"]["chunk"].content
                if content:
                    yield {"token": content}
        yield {"done": True}

    return token_stream()

app.run(port=8080)

import asyncio, sys
from kiboup import KiboAgentClient

async def chat_loop():
    async with KiboAgentClient(
        "http://localhost:8080",
        api_key="sk-chat-abc"
    ) as client:
        while True:
            user_input = input("You: ")
            sys.stdout.write("AI: ")

            async for chunk in client.stream(
                {"prompt": user_input}
            ):
                token = chunk.get("token")
                if token:
                    sys.stdout.write(token)
                    sys.stdout.flush()
            print("\n")

asyncio.run(chat_loop())

from kiboup import KiboAgentMcp

app = KiboAgentMcp(
    name="LangGraph MCP Server",
    api_keys={"sk-mcp-abc": "web-app"},
)

@app.tool()
async def ask(question: str) -> str:
    """Ask a question to GPT-4o-mini."""
    result = await graph.ainvoke(
        {"messages": [{"role": "user",
                      "content": question}]}
    )
    return result["messages"][-1].content

@app.tool()
def summarize(text: str) -> str:
    """Summarize text using GPT-4o-mini."""
    result = graph.invoke(
        {"messages": [{"role": "user",
         "content": f"Summarize:\n{text}"}]}
    )
    return result["messages"][-1].content

app.run(transport="sse")

import asyncio
from kiboup import KiboMcpClient

async def main():
    async with KiboMcpClient(
        "http://localhost:8000/sse",
        api_key="sk-mcp-abc"
    ) as client:
        tools = await client.list_tools()
        print(f"Tools: {tools}")

        result = await client.call_tool(
            "ask",
            {"question": "What is AI?"}
        )
        print(result)

asyncio.run(main())

from kiboup.a2a.server import (
    AgentExecutor, AgentSkill,
    KiboAgentA2A, TaskUpdater
)

app = KiboAgentA2A(
    name="Chat Agent",
    description="GPT-4o-mini via LangGraph",
    api_keys={"sk-a2a-xyz": "client"},
    skills=[AgentSkill(
        id="chat", name="Chat",
        description="Answer questions",
        tags=["chat", "qa"],
        input_modes=["text/plain"],
        output_modes=["text/plain"],
    )],
)

@app.executor
class ChatAgent(AgentExecutor):
    async def execute(self, ctx, queue):
        from a2a.utils import new_agent_text_message
        user_input = ctx.get_user_input()
        result = await graph.ainvoke(
            {"messages": [{"role": "user",
                          "content": user_input}]}
        )
        await queue.enqueue_event(
            new_agent_text_message(
                result["messages"][-1].content
            )
        )

    async def cancel(self, ctx, queue):
        updater = TaskUpdater(queue, ctx.task_id, ctx.context_id)
        await updater.cancel()

app.run()

import asyncio
from kiboup import KiboA2AClient

async def main():
    async with KiboA2AClient(
        "http://localhost:8000",
        api_key="sk-a2a-xyz"
    ) as client:
        card = client.agent_card
        print(f"Agent: {card.name}")
        print(f"Skills: {[s.name for s in card.skills]}")

        response = await client.send(
            "What is the capital of France?"
        )
        print(response)

asyncio.run(main())

// GET /.well-known/agent.json
{
  "name": "Chat Agent",
  "description": "GPT-4o-mini via LangGraph",
  "skills": [{
    "id": "chat",
    "name": "Chat",
    "tags": ["chat", "qa"]
  }]
}

from kiboup import KiboAgentApp
from kiboup.studio import StudioClient

app = KiboAgentApp()
studio = StudioClient(
    studio_url="http://127.0.0.1:8000",
    agent_id="researcher",
    agent_name="researcher",
    agent_endpoint="http://127.0.0.1:8081",
    capabilities=["research", "delegate"],
)
app.attach_studio(studio)

@app.entrypoint
async def invoke(payload, context):
    research = await do_research(payload["prompt"])

    # Feature flag: delegate to writer?
    if await studio.is_flag_enabled("delegate_to_writer"):
        # Discover writer agent via Studio
        agents = await studio.list_agents()
        writer = next(
            (a for a in agents
             if a["agent_id"] == "writer"), None
        )
        if writer:
            style = await studio.get_param(
                "writer_style", default="markdown"
            )
            # Delegate to writer endpoint
            result = await call_writer(
                writer["endpoint"], research, style
            )
            return {"response": result, "delegated": True}

    return {"response": research}

app.run(port=8081)

from kiboup import KiboAgentApp
from kiboup.studio import StudioClient

app = KiboAgentApp()
studio = StudioClient(
    studio_url="http://127.0.0.1:8000",
    agent_id="writer",
    agent_name="writer",
    agent_endpoint="http://127.0.0.1:8082",
    capabilities=["writing", "formatting"],
)
app.attach_studio(studio)

@app.entrypoint
async def invoke(payload, context):
    prompt = payload.get("prompt", "")
    research = payload.get("research", "")
    style = payload.get("style", "markdown")

    content = (
        f"Topic: {prompt}\n"
        f"Research:\n{research}\n"
        f"Format in {style} style."
    )

    result = await graph.ainvoke(
        {"messages": [{"role": "user",
                      "content": content}]}
    )
    return {
        "response": result["messages"][-1].content,
        "agent": "writer",
        "style": style,
    }

app.run(port=8082)

# Terminal 1: KiboStudio
uv run python examples/studio_example.py

# Terminal 2: Researcher
OPENAI_API_KEY=sk-... uv run python \
  examples/multi_agent_example.py --agent researcher

# Terminal 3: Writer
OPENAI_API_KEY=sk-... uv run python \
  examples/multi_agent_example.py --agent writer

from kiboup import KiboAgentApp

app = KiboAgentApp()

@app.entrypoint
async def invoke(payload, context):
    return {"response": "Hello from mTLS!"}

# Auto-generates CA, server & client certs
app.run(host="0.0.0.0", port=8443, mtls=True)

import asyncio
from kiboup import KiboAgentClient

async def main():
    async with KiboAgentClient(
        base_url="https://localhost:8443",
        mtls=True,
    ) as client:
        result = await client.invoke({"prompt": "Hello!"})

asyncio.run(main())