mirror of
https://github.com/hanzoai/agents.git
synced 2026-10-30 18:48:36 +00:00
zeekayandGitHub
5b5cf9e9db
Transforms hanzoai/agents from a marketplace config repo into a full agent orchestration platform by merging the AgentField infrastructure: - Go control plane (K8s-like agent orchestration, service mesh, DAG workflows) - Python SDK (FastAPI-based agent framework, LiteLLM, memory, DID/VC) - TypeScript SDK (@hanzo/agents - Vercel AI SDK, MCP, workflows) - Go SDK (idiomatic agent builder + client) - React 19 embedded dashboard (XyFlow DAGs, Recharts, Radix UI) - Docker/K8s/Helm deployment configs - Examples: hello world, deep research, RAG, simulation, benchmarks - Functional test suite across Python/Go/TypeScript All references rebranded: AgentField → Hanzo Agents - Module: github.com/hanzoai/agents/control-plane - Python: hanzo_agents - TypeScript: @hanzo/agents - Go: github.com/hanzoai/agents/sdk/go - Env vars: HANZO_AGENTS_* - Config: hanzo-agents.yaml Previous marketplace configs moved to config/ directory.
91 lines
2.5 KiB
Python
91 lines
2.5 KiB
Python
from types import SimpleNamespace
|
|
|
|
import httpx
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
|
|
from hanzo_agents.dynamic_skills import DynamicMCPSkillManager
|
|
|
|
|
|
class StubMCPClient:
|
|
def __init__(self, tools, *, result=None):
|
|
self._tools = tools
|
|
self._result = result or {"echo": "ok"}
|
|
self.calls = []
|
|
|
|
async def health_check(self):
|
|
return True
|
|
|
|
async def list_tools(self):
|
|
return self._tools
|
|
|
|
async def call_tool(self, name, args):
|
|
self.calls.append((name, args))
|
|
return self._result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dynamic_skill_registration(monkeypatch):
|
|
app = FastAPI()
|
|
app.node_id = "agent"
|
|
app.reasoners = []
|
|
app.skills = []
|
|
app.dev_mode = False
|
|
app.mcp_client_registry = SimpleNamespace(
|
|
clients={
|
|
"server": StubMCPClient(
|
|
tools=[
|
|
{
|
|
"name": "Echo",
|
|
"description": "Echo tool",
|
|
"inputSchema": {
|
|
"properties": {"text": {"type": "string"}},
|
|
"required": ["text"],
|
|
},
|
|
}
|
|
]
|
|
)
|
|
},
|
|
get_client=lambda alias: None,
|
|
)
|
|
|
|
def get_client(alias):
|
|
return app.mcp_client_registry.clients[alias]
|
|
|
|
app.mcp_client_registry.get_client = get_client
|
|
|
|
manager = DynamicMCPSkillManager(app)
|
|
await manager.discover_and_register_all_skills()
|
|
|
|
assert "server_Echo" in manager.registered_skills
|
|
assert any(skill["id"] == "server_Echo" for skill in app.skills)
|
|
|
|
client_stub = app.mcp_client_registry.get_client("server")
|
|
|
|
async with httpx.AsyncClient(
|
|
transport=httpx.ASGITransport(app=app), base_url="http://test"
|
|
) as client:
|
|
resp = await client.post(
|
|
"/skills/server_Echo",
|
|
json={"text": "hello"},
|
|
headers={"x-workflow-id": "wf", "x-execution-id": "exec"},
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
assert client_stub.calls[0][0].lower() == "echo"
|
|
assert client_stub.calls[0][1] == {"text": "hello"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mcp_registry_absent_succeeds_quickly():
|
|
app = FastAPI()
|
|
app.node_id = "agent"
|
|
app.reasoners = []
|
|
app.skills = []
|
|
app.dev_mode = False
|
|
app.mcp_client_registry = None
|
|
|
|
manager = DynamicMCPSkillManager(app)
|
|
await manager.discover_and_register_all_skills()
|
|
assert manager.registered_skills == {}
|