kiboserve (D10 S0 V Sk Y Oss)

KiboUP

用于通过HTTP、A2A和MCP部署AI代理的框架无关库 --通过以下方式内置可观测性、即时管理、评估和代理发现 Kibo工作室.

  

______________________________________________________________________

目录

• 概述
• 安装
• 协议：何时使用什么
• 快速开始

- HTTP代理（服务器+客户端） - 流媒体（SSE） - MCP服务器+客户端 - A2A服务器+客户端 - Chainlit聊天界面 - mTLS（双向TLS）

• Kibo工作室

- 入门指南 - 代理发现和多代理协作 - 痕迹和可观察性 - 图形可视化 - 聊天界面 - 功能标志和参数 - 提示管理 - 评估（法学硕士作为评委） - StudioClient SDK

• 例子

______________________________________________________________________

概述

KiboUP允许您使用以下工具构建和部署AI代理 一个代码库 并通过三种行业标准协议公开它们：

所有三个协议共享：

• API密钥认证中间件
• 结构化JSON日志记录 LLMUsage 元数据
• 健康检查和任务管理
• 可选的 Kibo工作室 可观测性集成

______________________________________________________________________

安装

# Core (HTTP only)
uv add kiboup

# With MCP support
uv add "kiboup[mcp]"

# With A2A support
uv add "kiboup[a2a]"

# With KiboStudio (observability, prompts, eval, discovery)
uv add "kiboup[studio]"

# Everything
uv add "kiboup[all]"

______________________________________________________________________

协议：何时使用什么

HTTP(KiboAgentApp / KiboAgentClient)

需要时使用HTTP 标准REST API 为了你的经纪人。这是最通用的选项——它适用于任何前端，支持通过SSE、WebSocket连接、任务跟踪进行流式传输，并与KiboStudio跟踪无缝集成。

最适合： Web应用程序、移动后端、微服务架构、任何使用HTTP的客户端。

特征：

• POST /invocations --调用代理
• GET /ping --健康检查（健康/忙碌）
• GET /tasks --列出活动任务
• DELETE /tasks/{id} --取消任务
• WS /ws --WebSocket端点
• SSE流媒体支持
• API密钥验证
• KiboStudio自动跟踪报告

MCP(KiboAgentMcp / KiboMcpClient)

当您的代理暴露时使用MCP 工具 其他代理或IDE可以发现并调用。模型上下文协议是基于工具的交互的标准，可以将其视为LLM的插件系统。

最适合： IDE集成（游标、VS代码）、基于工具的代理、将功能作为可调用函数公开的代理。

特征：

• 工具注册通过 @app.tool() 装饰器
• 资源和快速注册
• SSE和stdio传输
• 与MCP Inspector和所有MCP客户端兼容
• API密钥验证

A2A(KiboAgentA2A / KiboA2AClient)

当代理需要时使用A2A 发现并相互交流 使用谷歌的代理到代理协议。每个代理人在以下网址发布一张代理人卡 /.well-known/agent.json 描述其技能。

最适合： 多代理系统、代理市场、跨组织代理沟通。

特征：

• 代理卡自动生成 /.well-known/agent.json
• 基于技能的路线
• 任务生命周期管理（创建、取消）
• 承载令牌和API密钥认证
• 与任何A2A客户端兼容

______________________________________________________________________

快速开始

HTTP代理（服务器+客户端）

服务器 (agent_server_example.py):

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

from kiboup import KiboAgentApp, LLMUsage

app = KiboAgentApp(
    api_keys={
        "sk-frontend-abc": "web-app",
        "sk-agent-xyz": "recommender-agent",
    }
)

llm = ChatOpenAI(model="gpt-4o-mini")
graph_builder = StateGraph(MessagesState)

def chatbot(state: MessagesState):
    return {"messages": [llm.invoke(state["messages"])]}

graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph = graph_builder.compile()

def _extract_llm_usage(ai_message) -> LLMUsage:
    usage_meta = getattr(ai_message, "usage_metadata", None) or {}
    resp_meta = getattr(ai_message, "response_metadata", {})
    return LLMUsage(
        model=resp_meta.get("model_name"),
        provider="openai",
        input_tokens=usage_meta.get("input_tokens"),
        output_tokens=usage_meta.get("output_tokens"),
        total_tokens=usage_meta.get("total_tokens"),
    )

@app.entrypoint
async def invoke(payload, context):
    prompt = payload.get("prompt", "")
    result = await graph.ainvoke({"messages": [{"role": "user", "content": prompt}]})
    last_message = result["messages"][-1]

    usage = _extract_llm_usage(last_message)
    context._llm_usage = usage

    return {
        "response": last_message.content,
        "called_by": context.client_id,
        "llm_usage": usage.to_dict(),
    }

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8080, reload=True)

客户 (agent_client_example.py):

import asyncio
from kiboup import KiboAgentClient

async def main():
    async with KiboAgentClient(
        base_url="http://127.0.0.1:8080",
        api_key="sk-frontend-abc",
    ) as client:
        health = await client.ping()
        print(f"Server health: {health}")

        result = await client.invoke({"prompt": "What is the capital of France?"})
        print(f"Response: {result['response']}")

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

卷曲测试：

curl -X POST http://127.0.0.1:8080/invocations \
    -H "Content-Type: application/json" \
    -H "X-API-Key: sk-frontend-abc" \
    -d '{"prompt": "What is the capital of France?"}'

______________________________________________________________________

流媒体（SSE）

服务器 (stream_server_example.py):

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

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

graph_builder = StateGraph(MessagesState)

def chatbot(state: MessagesState):
    return {"messages": [llm.invoke(state["messages"])]}

graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph = graph_builder.compile()

@app.entrypoint
async def invoke(payload, context):
    prompt = payload.get("prompt", "")
    messages = payload.get("messages", [{"role": "user", "content": 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()

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8080)

客户 (stream_client_example.py):

import asyncio, sys
from kiboup import KiboAgentClient

async def chat_loop():
    async with KiboAgentClient("http://localhost:8080", api_key="sk-chat-abc") as client:
        health = await client.ping()
        print(f"Connected ({health['status']})\n")

        while True:
            user_input = input("You: ")
            if not user_input.strip():
                continue

            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")

if __name__ == "__main__":
    asyncio.run(chat_loop())

______________________________________________________________________

MCP服务器+客户端

服务器 (mcp_server_example.py):

import logging
from langchain_openai import ChatOpenAI
from langgraph.graph import START, MessagesState, StateGraph
from kiboup import KiboAgentMcp, LLMUsage

llm = ChatOpenAI(model="gpt-4o-mini")
graph_builder = StateGraph(MessagesState)

def chatbot(state: MessagesState):
    return {"messages": [llm.invoke(state["messages"])]}

graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph = graph_builder.compile()

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 the LangGraph agent powered by 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 the given text using GPT-4o-mini."""
    result = graph.invoke(
        {"messages": [{"role": "user", "content": f"Summarize this text:\n\n{text}"}]}
    )
    return result["messages"][-1].content

if __name__ == "__main__":
    app.run(transport="sse")

客户 (mcp_client_example.py):

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"Available tools: {tools}")

        result = await client.call_tool("ask", {"question": "What is the capital of France?"})
        print(f"Result: {result}")

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

______________________________________________________________________

A2A服务器+客户端

服务器 (a2a_server_example.py):

import logging
from langchain_openai import ChatOpenAI
from langgraph.graph import START, MessagesState, StateGraph
from kiboup import LLMUsage
from kiboup.a2a.server import AgentExecutor, AgentSkill, KiboAgentA2A, TaskUpdater

llm = ChatOpenAI(model="gpt-4o-mini")
graph_builder = StateGraph(MessagesState)

def chatbot(state: MessagesState):
    return {"messages": [llm.invoke(state["messages"])]}

graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph = graph_builder.compile()

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

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

    async def cancel(self, context, event_queue):
        updater = TaskUpdater(event_queue, context.task_id, context.context_id)
        await updater.cancel()

if __name__ == "__main__":
    app.run()

客户 (a2a_client_example.py):

import asyncio
from kiboup import KiboA2AClient

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

        response = await client.send("What is the capital of France?")
        print(f"Response: {response}")

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

______________________________________________________________________

Chainlit聊天界面

import chainlit as cl
from kiboup import KiboAgentClient

SERVER_URL = "http://localhost:8080"
API_KEY = "sk-chat-abc"

@cl.on_chat_start
async def on_start():
    cl.user_session.set("history", [])

@cl.on_message
async def on_message(message: cl.Message):
    history = cl.user_session.get("history", [])
    history.append({"role": "user", "content": message.content})

    response = cl.Message(content="")
    await response.send()

    full_response = ""
    async with KiboAgentClient(SERVER_URL, api_key=API_KEY) as client:
        async for chunk in client.stream({
            "prompt": message.content,
            "messages": history,
        }):
            token = chunk.get("token")
            if token:
                full_response += token
                await response.stream_token(token)

    await response.update()
    history.append({"role": "assistant", "content": full_response})
    cl.user_session.set("history", history)

首先启动流媒体服务器，然后运行：

uv run chainlit run examples/chainlit_example.py

______________________________________________________________________

mTLS（双向TLS）

KiboUP支持 自动双向TLS 对于所有协议。证书在首次运行时自动生成，并在到期前自动续订。

服务器：

from kiboup import KiboAgentApp

app = KiboAgentApp()

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

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!"})
        print(result["response"])

asyncio.run(main())

自定义证书目录 通过环境变量：

KIBO_CERTS_DIR=/path/to/certs uv run python my_server.py

自定义配置 通过 MTLSConfig:

from kiboup import MTLSConfig

config = MTLSConfig(
    certs_dir="/custom/certs",
    hostname="myagent.example.com",
    validity_days=365,
    renew_before_days=30,
)

app.run(port=8443, mtls=config)

证书存储在 ~/.kiboserve/certs/ 默认情况下。CA证书有效期为10年；服务器和客户端证书有效期为1年，到期前30天自动续订。

______________________________________________________________________

Kibo工作室

KiboStudio是内置的 开发者控制台 用于可观察性、即时管理、评估和代理发现。它作为一个具有SQLite后端的独立web服务器运行。

入门指南

from kiboup.studio import KiboStudio

studio = KiboStudio(db_path="kibostudio.db", debug=True)

if __name__ == "__main__":
    studio.run(host="0.0.0.0", port=8000, reload=True)

打开 http://127.0.0.1:8000 在您的浏览器中。

代理发现和多代理协作

KiboStudio充当 服务注册中心 代理在其中注册自己、发送心跳并在运行时发现彼此。

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)

一旦注册，代理商可以发现彼此：

agents = await studio.list_agents()
writer = next((a for a in agents if a.get("agent_id") == "writer"), None)
endpoint = writer["endpoint"]

这 发现 UI中的选项卡显示了所有已注册的代理的健康状态、正常运行时间、内存使用情况和功能。

痕迹和可观察性

每次调用通过 KiboAgentApp 附有 StudioClient 自动报告跟踪：

• 跨度层次结构：调用>代理运行>llm_call/tool_call/requiry
• 输入输出数据 对于每个跨度
• LLM令牌使用情况：模型、提供者、输入/输出/总令牌
• 持续时间 和 状态 （正常/错误）
• 属性：自定义键值对

这 痕迹 选项卡组按代理进行跟踪，并一目了然地显示时间、状态和令牌消耗。

图形可视化

这 图 选项卡呈现每个跟踪跨度层次结构的ADK样式的可视化图形：

• 代理节点：带机器人表情符号的绿色填充椭圆
• 工具节点：圆形矩形，带扳手表情符号
• LLM节点：带大脑的圆形矩形表情符号
• 检索节点：带有放大镜表情符号的圆形矩形
• 深色背景(#333537)，从左到右布局，带箭头的贝塞尔曲线边
• 单击任何节点以检查其跨度详细信息

聊天界面

这 聊天 选项卡提供了一个内置的聊天界面，可以直接从浏览器测试任何注册的代理。选择一个代理，键入一条消息，然后查看使用完整markdown支持呈现的响应。

功能标志和参数

无需重新部署即可在运行时控制代理行为：

功能开关 --打开/关闭功能：

delegate_enabled = await studio.is_flag_enabled("delegate_to_writer")
if not delegate_enabled:
    return {"response": research, "delegated": False}

参数 --动态配置值：

writer_style = await studio.get_param("writer_style", default="markdown")

支持 全球 （适用于所有代理人）以及 每个代理人 范围。SDK缓存具有30秒TTL的值以提高性能。

提示管理

这 提示 选项卡允许您通过以下方式管理提示模板：

• 版本历史
• 变量提取
• 活动版本选择
• 每个版本的模型配置

代理可以在运行时获取提示：

prompt = await studio.get_prompt("research_system_prompt")
content = prompt["content"]

评估（法学硕士作为评委）

这 评估 选项卡使用LLM作为判断方法（GPT-4o-mini）对痕迹进行自动质量评估。它根据四个指标对每条跟踪进行评分：

每个指标的得分为0.0到1.0。结果按轨迹存储，并用视觉评分条显示。

StudioClient SDK

这 StudioClient 提供了一个完整的异步Python SDK，供代理与KiboStudio交互：

from kiboup.studio import StudioClient

studio = StudioClient(
    studio_url="http://127.0.0.1:8000",
    agent_id="my-agent",
    agent_name="My Agent",
    agent_endpoint="http://127.0.0.1:8081",
    capabilities=["chat"],
    heartbeat_interval_s=15,
)

async with studio:
    # Discovery
    agents = await studio.list_agents()

    # Feature flags
    enabled = await studio.is_flag_enabled("my_flag")

    # Parameters
    value = await studio.get_param("my_param", default="fallback")

    # Prompts
    prompt = await studio.get_prompt("system_prompt")

    # Traces
    await studio.send_traces(trace_data)

客户端也可以直接嵌入到 KiboAgentClient 或 KiboMcpClient:

async with KiboAgentClient(
    base_url="http://localhost:8080",
    studio_url="http://localhost:8000",
    agent_id="my-agent",
) as client:
    result = await client.invoke({"prompt": "Hello"})
    flags = await client.studio.get_flags()

______________________________________________________________________

例子

运行任何示例：

OPENAI_API_KEY=sk-... uv run python examples/.py

______________________________________________________________________

许可证

麻省理工学院