Skip to main content

Custom Agent (Python / TypeScript)

Build your own MCP-aware agent against ColabHive. Use this if you're rolling a custom orchestrator or embedding MCP in an existing app.

Two transport paths:

  • Hosted (SSE / HTTP) — connect to https://mcp.colabhive.com/mcp. Zero install. Recommended.
  • Local stdio — launch colabhive-mcp as a child process. Key never leaves the box.

Python — talking to the hosted endpoint​

Plain httpx, no MCP SDK needed:

"""mcp_hosted.py — minimal client against mcp.colabhive.com."""
import os
import httpx

KEY = os.environ["COLABHIVE_API_KEY"]
URL = "https://mcp.colabhive.com/mcp"
HEADERS = {"X-API-Key": KEY, "Content-Type": "application/json"}


def rpc(method: str, params: dict | None = None, id_: int = 1) -> dict:
    body = {"jsonrpc": "2.0", "id": id_, "method": method}
    if params is not None:
        body["params"] = params
    r = httpx.post(URL, headers=HEADERS, json=body, timeout=60)
    r.raise_for_status()
    return r.json()


# Discover
print("Initializing...")
rpc("initialize", {
    "protocolVersion": "2026-01-01",
    "capabilities": {},
    "clientInfo": {"name": "my-app", "version": "0.1"},
})

print("Tools:")
listed = rpc("tools/list", id_=2)
for t in listed["result"]["tools"][:5]:
    print(f"  - {t['name']}: {t['description'][:60]}")

# Invoke an LLM
print("\nCalling qwen...")
result = rpc("tools/call", {
    "name": "qwen-2.5-7b-instruct-public",
    "arguments": {"messages": [{"role": "user", "content": "Say hello in 5 languages."}]},
}, id_=3)
print(result["result"]["content"][0]["text"])

Run:

COLABHIVE_API_KEY=hive_xxx python mcp_hosted.py

Python — using the official MCP SDK against the hosted endpoint​

pip install mcp httpx
import asyncio, os
from mcp.client.sse import sse_client
from mcp.client.session import ClientSession


async def main():
    headers = {"X-API-Key": os.environ["COLABHIVE_API_KEY"]}
    async with sse_client("https://mcp.colabhive.com/mcp", headers=headers) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            tools = await session.list_tools()
            print(f"{len(tools.tools)} tools")

            result = await session.call_tool(
                "embeddings-public",
                arguments={"text": "hello world"},
            )
            print(result.content)


asyncio.run(main())

Python — local stdio​

For privacy-sensitive setups where the key shouldn't traverse the network beyond api.colabhive.com:

import asyncio, os
from mcp.client.stdio import StdioServerParameters, stdio_client
from mcp.client.session import ClientSession


async def main():
    server = StdioServerParameters(
        command="uvx",
        args=["colabhive-mcp@latest"],
        env={"COLABHIVE_API_KEY": os.environ["COLABHIVE_API_KEY"]},
    )
    async with stdio_client(server) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            tools = await session.list_tools()
            ...

asyncio.run(main())

TypeScript — @modelcontextprotocol/sdk​

npm i @modelcontextprotocol/sdk

Hosted (SSE)​

import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { SSEClientTransport } from "@modelcontextprotocol/sdk/client/sse.js";

const transport = new SSEClientTransport(
  new URL("https://mcp.colabhive.com/mcp"),
  { requestInit: { headers: { "X-API-Key": process.env.COLABHIVE_API_KEY! } } },
);

const client = new Client({ name: "my-app", version: "1.0.0" }, { capabilities: {} });
await client.connect(transport);

const { tools } = await client.listTools();
console.log(`${tools.length} ColabHive tools`);

const result = await client.callTool({
  name: "embeddings-public",
  arguments: { text: "hello world" },
});
console.log(result.content);

Local stdio​

import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";

const transport = new StdioClientTransport({
  command: "uvx",
  args: ["colabhive-mcp@latest"],
  env: { COLABHIVE_API_KEY: process.env.COLABHIVE_API_KEY! },
});
const client = new Client(/* ... */);
await client.connect(transport);

Driving an LLM with MCP tools​

The common pattern is: surface MCP tools to Claude / GPT, let the model decide which to call:

import asyncio, os
import anthropic
from mcp.client.sse import sse_client
from mcp.client.session import ClientSession

claude = anthropic.Anthropic()  # ANTHROPIC_API_KEY in env


async def run():
    headers = {"X-API-Key": os.environ["COLABHIVE_API_KEY"]}
    async with sse_client("https://mcp.colabhive.com/mcp", headers=headers) as (r, w):
        async with ClientSession(r, w) as mcp:
            await mcp.initialize()
            tools_resp = await mcp.list_tools()

            anthropic_tools = [
                {
                    "name": t.name.replace("-", "_"),  # Anthropic disallows '-'
                    "description": t.description,
                    "input_schema": t.inputSchema,
                }
                for t in tools_resp.tools
            ]
            slug_map = {t.name.replace("-", "_"): t.name for t in tools_resp.tools}

            msg = claude.messages.create(
                model="claude-opus-4-7",
                max_tokens=4096,
                tools=anthropic_tools,
                messages=[{"role": "user", "content": "Embed 'hello world' and show the first 5 dims."}],
            )

            for block in msg.content:
                if block.type == "tool_use":
                    result = await mcp.call_tool(slug_map[block.name], arguments=block.input)
                    print(result.content)


asyncio.run(run())

Embedding MCP in your own framework​

The protocol is JSON-RPC 2.0 over HTTP or stdio. Minimal methods:

  • initialize — handshake
  • tools/list — discover
  • tools/call — invoke

See the MCP spec for full message shapes.

Skip MCP — direct Actions API​

If MCP itself is overkill, hit the Actions API directly. Same shape, same auth, no JSON-RPC layer.

import httpx, os
BASE = "https://api.colabhive.com/api/builder/v1"
KEY = os.environ["COLABHIVE_API_KEY"]

r = httpx.get(f"{BASE}/actions?kind=llm", headers={"X-API-Key": KEY})
print(r.json())

r = httpx.post(
    f"{BASE}/actions/qwen-2.5-7b-instruct-public:invoke",
    headers={"X-API-Key": KEY},
    json={"input": {"messages": [{"role": "user", "content": "Hi"}]}},
)
print(r.json()["result"])

See also​