उदाहरण

# .env
DEDALUS_API_KEY=dsk-live-...

# Supabase
SUPABASE_URL=https://xxx.supabase.co
SUPABASE_SECRET_KEY=eyJ...

# GitHub
GITHUB_TOKEN=ghp_...
GITHUB_BASE_URL=https://api.github.com

# server.py
import os
from dedalus_mcp import MCPServer
from dedalus_mcp.server import TransportSecuritySettings
from dedalus_mcp.auth import Connection, SecretKeys

# Connection परिभाषित करें (क्रेडेंशियल रनटाइम पर client द्वारा प्रदान किए जाते हैं)
github = Connection(
    name="github",
    secrets=SecretKeys(token="GITHUB_TOKEN"),
    auth_header_format="token {api_key}"
)

supabase = Connection(
    name="supabase",
    secrets=SecretKeys(token="SUPABASE_SECRET_KEY"),
    auth_header_format="Bearer {api_key}"
)

def create_server() -> MCPServer:
    return MCPServer(
        name="example-dedalus-mcp",
        connections=[github, supabase],
        http_security=TransportSecuritySettings(enable_dns_rebinding_protection=False),
        streamable_http_stateless=True,
        authorization_server=os.getenv("DEDALUS_AS_URL", "https://as.dedaluslabs.ai"),
    )

# gh.py
from dataclasses import dataclass
from typing import Any
from dedalus_mcp import HttpMethod, HttpRequest, get_context, tool
from dedalus_mcp.types import ToolAnnotations

@dataclass(frozen=True)
class GhResult:
    """GitHub API परिणाम।"""
    success: bool
    data: Any = None
    error: str | None = None

@dataclass(frozen=True)
class GhUser:
    """GitHub उपयोगकर्ता प्रोफ़ाइल।"""
    login: str
    name: str | None
    bio: str | None
    public_repos: int
    followers: int

@dataclass(frozen=True)
class GhRepo:
    """GitHub रिपॉज़िटरी सारांश।"""
    name: str
    full_name: str
    stars: int
    language: str | None
    updated_at: str

async def _gh_request(method: HttpMethod, path: str, body: Any = None) -> GhResult:
    """GitHub API अनुरोध निष्पादित करें।"""
    ctx = get_context()
    resp = await ctx.dispatch("github", HttpRequest(method=method, path=path, body=body))
    if resp.success:
        return GhResult(success=True, data=resp.response.body)
    return GhResult(success=False, error=resp.error.message if resp.error else "Request failed")

@tool(
    description="प्रमाणित GitHub उपयोगकर्ता की प्रोफ़ाइल प्राप्त करें",
    tags=["user", "read"],
    annotations=ToolAnnotations(readOnlyHint=True),
)
async def gh_whoami() -> GhResult:
    result = await _gh_request(HttpMethod.GET, "/user")
    if result.success and result.data:
        user = GhUser(
            login=result.data["login"],
            name=result.data.get("name"),
            bio=result.data.get("bio"),
            public_repos=result.data.get("public_repos", 0),
            followers=result.data.get("followers", 0),
        )
        return GhResult(success=True, data=user)
    return result

@tool(
    description="प्रमाणित उपयोगकर्ता की रिपॉज़िटरी सूचीबद्ध करें",
    tags=["repos", "read"],
    annotations=ToolAnnotations(readOnlyHint=True),
)
async def gh_list_repos(per_page: int = 10) -> GhResult:
    result = await _gh_request(HttpMethod.GET, f"/user/repos?per_page={per_page}&sort=updated")
    if result.success and isinstance(result.data, list):
        repos = [
            GhRepo(
                name=r["name"],
                full_name=r["full_name"],
                stars=r.get("stargazers_count", 0),
                language=r.get("language"),
                updated_at=r.get("updated_at", ""),
            )
            for r in result.data
        ]
        return GhResult(success=True, data=repos)
    return result

# .env
X_BEARER_TOKEN=AAAA...  # App-only bearer token
X_API_KEY=...           # Optional: for user-context endpoints
X_API_KEY_SECRET=...    # वैकल्पिक: उपयोगकर्ता-संदर्भ एंडपॉइंट्स के लिए

# x.py
from dataclasses import dataclass
from typing import Any
from dedalus_mcp import HttpMethod, HttpRequest, get_context, tool
from dedalus_mcp.auth import Connection, SecretKeys
from dedalus_mcp.types import ToolAnnotations

x = Connection(
    name="x",
    secrets=SecretKeys(token="X_BEARER_TOKEN"),
    auth_header_format="Bearer {api_key}",
)

DEFAULT_TWEET_FIELDS = "id,text,author_id,created_at,public_metrics"
DEFAULT_USER_FIELDS = "id,name,username,description,public_metrics"

@dataclass(frozen=True)
class XResult:
    """X API result."""
    success: bool
    data: Any = None
    meta: dict | None = None
    error: str | None = None

@dataclass(frozen=True)
class XUser:
    """X user profile."""
    id: str
    name: str
    username: str
    description: str | None
    followers_count: int
    following_count: int

@dataclass(frozen=True)
class XTweet:
    """X tweet."""
    id: str
    text: str
    author_id: str
    created_at: str
    retweet_count: int
    like_count: int

async def _x_request(path: str) -> XResult:
    """Execute X API request."""
    ctx = get_context()
    resp = await ctx.dispatch("x", HttpRequest(method=HttpMethod.GET, path=path))
    if resp.success:
        body = resp.response.body or {}
        return XResult(success=True, data=body.get("data"), meta=body.get("meta"))
    return XResult(success=False, error=resp.error.message if resp.error else "अनुरोध विफल रहा")

@tool(
    description="उपयोगकर्ता नाम से X उपयोगकर्ता प्राप्त करें",
    tags=["user", "read"],
    annotations=ToolAnnotations(readOnlyHint=True),
)
async def x_get_user_by_username(username: str) -> XResult:
    result = await _x_request(f"/2/users/by/username/{username}?user.fields={DEFAULT_USER_FIELDS}")
    if result.success and result.data:
        metrics = result.data.get("public_metrics", {})
        user = XUser(
            id=result.data["id"],
            name=result.data["name"],
            username=result.data["username"],
            description=result.data.get("description"),
            followers_count=metrics.get("followers_count", 0),
            following_count=metrics.get("following_count", 0),
        )
        return XResult(success=True, data=user)
    return result

@tool(
    description="हाल के ट्वीट खोजें (पिछले 7 दिन)",
    tags=["search", "read"],
    annotations=ToolAnnotations(readOnlyHint=True),
)
async def x_search_recent(query: str, max_results: int = 10) -> XResult:
    from urllib.parse import quote
    max_results = max(10, min(100, max_results))
    result = await _x_request(
        f"/2/tweets/search/recent?query={quote(query)}&tweet.fields={DEFAULT_TWEET_FIELDS}&max_results={max_results}"
    )
    if result.success and result.data:
        tweets = [
            XTweet(
                id=t["id"],
                text=t["text"],
                author_id=t["author_id"],
                created_at=t.get("created_at", ""),
                retweet_count=t.get("public_metrics", {}).get("retweet_count", 0),
                like_count=t.get("public_metrics", {}).get("like_count", 0),
            )
            for t in result.data
        ]
        return XResult(success=True, data=tweets, meta=result.meta)
    return result

@tool(
    description="किसी उपयोगकर्ता के हाल के ट्वीट प्राप्त करें",
    tags=["tweet", "read"],
    annotations=ToolAnnotations(readOnlyHint=True),
)
async def x_get_user_tweets(user_id: str, max_results: int = 10) -> XResult:
    max_results = max(5, min(100, max_results))
    result = await _x_request(
        f"/2/users/{user_id}/tweets?tweet.fields={DEFAULT_TWEET_FIELDS}&max_results={max_results}"
    )
    if result.success and result.data:
        tweets = [
            XTweet(
                id=t["id"],
                text=t["text"],
                author_id=t["author_id"],
                created_at=t.get("created_at", ""),
                retweet_count=t.get("public_metrics", {}).get("retweet_count", 0),
                like_count=t.get("public_metrics", {}).get("like_count", 0),
            )
            for t in result.data
        ]
        return XResult(success=True, data=tweets, meta=result.meta)
    return result

# .env
OAUTH_ENABLED=true
OAUTH_AUTHORIZE_URL=https://accounts.google.com/o/oauth2/auth
OAUTH_TOKEN_URL=https://oauth2.googleapis.com/token
OAUTH_CLIENT_ID=your-client-id.apps.googleusercontent.com
OAUTH_CLIENT_SECRET=your-client-secret
OAUTH_SCOPES_AVAILABLE=https://www.googleapis.com/auth/gmail.readonly,https://www.googleapis.com/auth/gmail.modify
OAUTH_BASE_URL=https://gmail.googleapis.com

DEDALUS_API_KEY=dsk-live-...
DEDALUS_API_URL=https://api.dedaluslabs.ai
DEDALUS_AS_URL=https://as.dedaluslabs.ai

# server.py
import os
from dedalus_mcp import MCPServer
from dedalus_mcp.server import TransportSecuritySettings
from gmail import gmail, gmail_tools

def create_server() -> MCPServer:
    return MCPServer(
        name="gmail-mcp",
        connections=[gmail],
        http_security=TransportSecuritySettings(enable_dns_rebinding_protection=False),
        streamable_http_stateless=True,
        authorization_server=os.getenv("DEDALUS_AS_URL", "https://as.dedaluslabs.ai"),
    )

async def main() -> None:
    server = create_server()
    server.collect(*gmail_tools)
    await server.serve(port=8080)

# gmail.py
import os
from dataclasses import dataclass
from typing import Any
from dedalus_mcp import HttpMethod, HttpRequest, get_context, tool
from dedalus_mcp.auth import Connection, OAuthConfig
from dedalus_mcp.types import ToolAnnotations

gmail = Connection(
    name="gmail",
    oauth=OAuthConfig(
        client_id=os.getenv("OAUTH_CLIENT_ID"),
        client_secret=os.getenv("OAUTH_CLIENT_SECRET"),
        authorize_url=os.getenv("OAUTH_AUTHORIZE_URL"),
        token_url=os.getenv("OAUTH_TOKEN_URL"),
        scopes=os.getenv("OAUTH_SCOPES_AVAILABLE", "").split(","),
    ),
    auth_header_format="Bearer {access_token}",
)

@dataclass(frozen=True)
class GmailResult:
    """Gmail API result."""
    success: bool
    data: Any = None
    error: str | None = None

@dataclass(frozen=True)
class GmailMessage:
    """Gmail message summary."""
    id: str
    thread_id: str
    snippet: str | None
    label_ids: list[str]

async def _gmail_request(method: HttpMethod, path: str) -> GmailResult:
    """Execute Gmail API request."""
    ctx = get_context()
    resp = await ctx.dispatch("gmail", HttpRequest(method=method, path=path))
    if resp.success:
        return GmailResult(success=True, data=resp.response.body)
    return GmailResult(success=False, error=resp.error.message if resp.error else "अनुरोध विफल रहा")

@tool(
    description="List recent emails",
    tags=["email", "read"],
    annotations=ToolAnnotations(readOnlyHint=True),
)
async def gmail_list_messages(max_results: int = 10, query: str = "") -> GmailResult:
    path = f"/gmail/v1/users/me/messages?maxResults={max_results}"
    if query:
        path += f"&q={query}"
    result = await _gmail_request(HttpMethod.GET, path)
    if result.success and result.data:
        messages = [
            GmailMessage(
                id=m["id"],
                thread_id=m.get("threadId", ""),
                snippet=None,
                label_ids=[],
            )
            for m in result.data.get("messages", [])
        ]
        return GmailResult(success=True, data=messages)
    return result

@tool(
    description="ID के आधार पर ईमेल विवरण प्राप्त करें",
    tags=["email", "read"],
    annotations=ToolAnnotations(readOnlyHint=True),
)
async def gmail_get_message(message_id: str) -> GmailResult:
    result = await _gmail_request(HttpMethod.GET, f"/gmail/v1/users/me/messages/{message_id}")
    if result.success and result.data:
        msg = result.data
        message = GmailMessage(
            id=msg["id"],
            thread_id=msg.get("threadId", ""),
            snippet=msg.get("snippet"),
            label_ids=msg.get("labelIds", []),
        )
        return GmailResult(success=True, data=message)
    return result

# Clone the repo
git clone https://github.com/dedalus-labs/example-dedalus-mcp
cd example-dedalus-mcp

# मुख्य उदाहरण के लिए (GitHub + Supabase)
cp .env.example .env
# Edit .env with your credentials

uv sync
uv run python src/main.py

# For X
cd x-mcp
cp .env.example .env
uv sync
uv run python src/main.py

# For Gmail
cd gmail-mcp
cp .env.example .env
uv sync
uv run python src/main.py