mirror of
https://github.com/CopilotKit/CopilotKit.git
synced 2026-09-14 16:26:20 +08:00
Mike Ryan
92afdb1f2c
Provide native Intelligence SDK operations and runtime routes with application-owned identity, durable runs, A2UI, MCP Apps, entitlements, Inspector metadata, and telemetry. Include package docs, hosting examples, and regression tests.
87 lines
3.0 KiB
Python
87 lines
3.0 KiB
Python
"""Test harness configuration only; all behavior lives in the installed library."""
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import socket
|
|
|
|
import uvicorn
|
|
|
|
from copilotkit_runtime import (
|
|
A2UIConfig,
|
|
HttpAgent,
|
|
IntelligenceRuntime,
|
|
MCPAppsConfig,
|
|
MCPServer,
|
|
RuntimeConfig,
|
|
Telemetry,
|
|
User,
|
|
)
|
|
|
|
|
|
async def main():
|
|
config = json.loads(os.environ["CPK_CONFIG"])
|
|
|
|
async def identify(request):
|
|
return User(
|
|
request.headers.get("x-test-user-id", "test-user"),
|
|
request.headers.get("x-test-user-name", "Test User"),
|
|
)
|
|
|
|
runtime = IntelligenceRuntime(
|
|
RuntimeConfig(
|
|
api_key=config["apiKey"],
|
|
api_url=config["apiUrl"],
|
|
runner_url=config["runnerUrl"],
|
|
client_url=config["clientUrl"],
|
|
),
|
|
agents={"default": HttpAgent(config["agentUrl"], description="Conformance agent")},
|
|
memory_policy=None
|
|
if config.get("omitMemoryPolicy")
|
|
else lambda *_: config.get("memoryGrant", {"user": "read-write", "project": "read-write"}),
|
|
identify_user=identify,
|
|
telemetry=Telemetry(
|
|
enabled=not config.get("telemetryDisabled", False),
|
|
sample_rate=config.get("telemetrySampleRate", 0.05),
|
|
telemetry_id=config.get("telemetryId"),
|
|
license_token=config.get("licenseToken"),
|
|
url=config.get("telemetryUrl", "https://telemetry.copilotkit.ai/ingest"),
|
|
),
|
|
a2ui=A2UIConfig(
|
|
enabled=config["a2ui"].get("enabled", True),
|
|
agents=tuple(config["a2ui"]["agents"]) if "agents" in config["a2ui"] else None,
|
|
schema=config["a2ui"].get("schema"),
|
|
inject_tool=config["a2ui"].get("injectA2UITool", False),
|
|
tool_names=tuple(config["a2ui"].get("a2uiToolNames", ["render_a2ui"])),
|
|
default_catalog_id=config["a2ui"].get("defaultCatalogId"),
|
|
max_attempts=config["a2ui"].get("recovery", {}).get("maxAttempts", 3),
|
|
show_progress_tokens=config["a2ui"].get("recovery", {}).get("showProgressTokens", True),
|
|
debug_exposure=config["a2ui"].get("recovery", {}).get("debugExposure"),
|
|
)
|
|
if config.get("a2ui") is not None
|
|
else None,
|
|
mcp_apps=MCPAppsConfig(
|
|
servers=tuple(
|
|
MCPServer(
|
|
url=server["url"],
|
|
server_id=server.get("serverId"),
|
|
agent_id=server.get("agentId"),
|
|
headers=server.get("headers", {}),
|
|
)
|
|
for server in config["mcpApps"].get("servers", [])
|
|
)
|
|
)
|
|
if config.get("mcpApps")
|
|
else None,
|
|
)
|
|
sock = socket.socket()
|
|
sock.bind(("127.0.0.1", config.get("port", 0)))
|
|
sock.listen(128)
|
|
print(json.dumps({"port": sock.getsockname()[1]}), flush=True)
|
|
server = uvicorn.Server(uvicorn.Config(runtime, log_level="warning", lifespan="on"))
|
|
await server.serve(sockets=[sock])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|