Client Overview

Interact programmatically with any MCP‑compatible server: local, remote, or in‑memory - using the fully‑async Client class shipped with Hyperia Protocol.

Introduced in v2.0 · design‑overhauled in v2.2 for richer streaming and multi‑server support.

---

1 Architecture

flowchart LR
    subgraph App
        direction TB
        A[Your Python Code] -->|await| C(Client)
    end
    C --> T[Transport]
    T -->|wire| S(Server)

• Client – constructs MCP requests (tools/call, resources/read…), tracks sessions, aggregates notifications.

• Transport – actual pipe (STDIO process, HTTP/2 stream, in‑mem RPC,…).

Separation of concerns means you can swap transports without touching business logic.

---

2 Transport Inference Cheatsheet

Input to Client( …)	Detected Transport	Typical Use‑Case
Hyperia() instance	InMemoryTransport	Unit tests, plugins
Path *.py	PythonStdioTransport	Spawn Python script
Path *.js	NodeStdioTransport	Node servers
http(s):// URL	StreamableHttpTransport	Cloud functions, micro‑services
MCPConfig dict/file	CompositeTransport	Multi‑server routing
Pre‑built Transport	As‑is	Full control (headers, env)

from hyperia import Client, Hyperia
backend = Hyperia("MyAPI")

Client(backend)                   # in‑mem
Client("api.py")                  # stdio
Client("https://api.example.com") # HTTP

Tip – instantiate transports directly when you need custom TLS certs, auth headers, or env vars.

---

3 Connecting to Many Servers (Composite Clients)

A single MCPConfig can declare multiple backends which the client auto‑mounts with name prefixes:

cfg = {
  "mcpServers": {
    "weather":  {"url": "https://weather.example.com/mcp"},
    "billing":  {"command": "python", "args": ["billing_server.py"]},
  }
}
client = Client(cfg)

• Tools → weather_get_forecast, billing_create_invoice

• Resources → weather://weather/icons/snow, resource://billing/plans/list

No additional routing code required—client picks server from the prefix.

---

4 Lifecycle ⇢ async with

async with Client("server.py") as c:
    avg = await c.call_tool("calc_avg", {"nums": [1,2,3]})

• __aenter__ – establishes transport, negotiates roots, sets heart‑beats.

• __aexit__ – gracefully closes session, kills child proc if STDIO.

You can reuse the same client object; reconnects automatically between context blocks.

---

5 High‑Level "Happy‑Path" API

Tools

tools = await c.list_tools()        # List[Tool]
res   = await c.call_tool("add", {"a":5,"b":7})
print(res[0].text)  # → "12"

Add timeout= or progress_handler= per call.

Resources

await c.list_resources()
content = await c.read_resource("data://weather/london")

Prompts

prompt = await c.get_prompt("email_draft", {"topic": "Launch"})
print("\n".join(m.text for m in prompt))

Ping / Health‑Check

await c.ping()

---

6 Raw Protocol Access

Every helper has an *_mcp twin that returns exact spec objects—future proof for breaking changes:

result = await c.call_tool_mcp("add", {"a":2,"b":2})
print(result.outputs[0].content)

---

7 Progress, Logs, Roots & Sampling

Attach handlers globally or per‑call:

from hyperia.client import ProgressHandler, LogHandler

class Spinner(ProgressHandler):
    async def on_progress(self, pct):
        print(f"\r{pct*100:.0f}%", end="")

async with Client("job.py", handlers=[LogHandler(print)]) as c:
    await c.call_tool("long_job", progress_handler=Spinner())

Handlers available:

• RootsHandler / RootsList – reacts to dynamic root updates.

• LogHandler – real‑time ctx.info() / ctx.error() streaming.

• ProgressHandler – percentage updates.

• SamplingHandler – (roadmap) client‑side LLM callbacks.

---

8 Timeout Strategy

• Global — Client(timeout=5)

• Per‑request — overrides via timeout= param.

• HTTP chooses the lower value; SSE honours per‑call value first.

---

9 Error Taxonomy

Exception	Trigger
ClientError	Server threw ToolError / ResourceError
ConnectionError	Transport connection failed
TimeoutError	Hard timeout hit

Inspect raw result for partial data:

res = await c.call_tool_mcp("maybe_fails", {...})
if res.isError:
    ...

---

10 Quick‑Ref Code Snippet

import asyncio, json
from hyperia import Client

async def main():
    async with Client("https://api.weather.example.com/mcp") as cli:
        wx = await cli.call_tool("get_forecast", {"city": "Tokyo"})
        print(json.loads(wx[0].text)["forecast"])

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