# Copyright (c) 2026 Dedalus Labs, Inc. and its contributors
# SPDX-License-Identifier: MIT
"""Sample MCP client demonstrating credential encryption and JIT token exchange.
Environment variables:
DEDALUS_API_KEY: Your Dedalus API key (dsk_*)
DEDALUS_API_URL: Product API base URL
DEDALUS_AS_URL: Authorization server URL (for encryption key)
X_BEARER_TOKEN: X/Twitter bearer token
"""
import asyncio
import os
from dotenv import load_dotenv
load_dotenv()
from dedalus_labs import AsyncDedalus, DedalusRunner
from dedalus_mcp.auth import Connection, SecretKeys, SecretValues
class MissingEnvError(ValueError):
"""Required environment variable not set."""
def get_env(key: str) -> str:
"""Get required env var or raise."""
val = os.getenv(key)
if not val:
raise MissingEnvError(key)
return val
API_URL = get_env("DEDALUS_API_URL")
AS_URL = get_env("DEDALUS_AS_URL")
DEDALUS_API_KEY = os.getenv("DEDALUS_API_KEY")
# Debug: print env vars
print("=== Environment ===")
print(f" DEDALUS_API_URL: {API_URL}")
print(f" DEDALUS_AS_URL: {AS_URL}")
print(f" DEDALUS_API_KEY: {DEDALUS_API_KEY[:20]}..." if DEDALUS_API_KEY else " DEDALUS_API_KEY: None")
# Connection: schema for X/Twitter API
x = Connection(
name="x",
secrets=SecretKeys(token="X_BEARER_TOKEN"),
base_url="https://api.x.com",
auth_header_format="Bearer {api_key}",
)
# SecretValues: binds actual credentials to a Connection schema.
# Encrypted client-side, decrypted in secure enclave at dispatch time.
x_secrets = SecretValues(x, token=os.getenv("X_BEARER_TOKEN", ""))
async def run_with_runner() -> None:
"""Demo using DedalusRunner (handles multi-turn, aggregates results)."""
client = AsyncDedalus(api_key=DEDALUS_API_KEY, base_url=API_URL, as_base_url=AS_URL)
runner = DedalusRunner(client)
result = await runner.run(
input="Get info about the Twitter user @elonmusk",
model="openai/gpt-4.1",
mcp_servers=["windsor/x-mcp"], # Update after deployment
credentials=[x_secrets],
)
print("=== Model Output ===")
print(result.output)
if result.mcp_results:
print("\n=== MCP Tool Results ===")
for r in result.mcp_results:
print(f" {r.tool_name} ({r.duration_ms}ms): {str(r.result)[:200]}")
async def run_raw() -> None:
"""Demo using raw client (single request, full control)."""
client = AsyncDedalus(api_key=DEDALUS_API_KEY, base_url=API_URL, as_base_url=AS_URL)
resp = await client.chat.completions.create(
model="openai/gpt-4.1",
messages=[
{
"role": "user",
"content": "Get info about the Twitter user @windsornguyen",
}
],
mcp_servers=["windsor/x-mcp"], # Update after deployment
credentials=[x_secrets],
)
print("=== Model Output ===")
print(resp.choices[0].message.content)
if resp.mcp_tool_results:
print("\n=== MCP Tool Results ===")
for r in resp.mcp_tool_results:
print(f" {r.tool_name} ({r.duration_ms}ms): {str(r.result)[:200]}")
async def main() -> None:
"""Run both demo modes."""
print("=" * 60)
print("DedalusRunner")
print("=" * 60)
await run_with_runner()
print("\n" + "=" * 60)
print("Raw Client")
print("=" * 60)
await run_raw()
if __name__ == "__main__":
asyncio.run(main())
