GitHub + Supabase
A multi-connection MCP server exposing GitHub and Supabase tools.Environment
Report incorrect code
Copy
Ask AI
# .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
Report incorrect code
Copy
Ask AI
# server.py
import os
from dedalus_mcp import MCPServer
from dedalus_mcp.server import TransportSecuritySettings
from dedalus_mcp.auth import Connection, SecretKeys
# Define connections (credentials provided by client at runtime)
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"),
)
GitHub tools
Report incorrect code
Copy
Ask AI
# gh.py
from dedalus_mcp import HttpMethod, HttpRequest, get_context, tool
from dedalus_mcp.types import ToolAnnotations
@tool(
description="Get the authenticated GitHub user's profile",
tags=["user", "read"],
annotations=ToolAnnotations(readOnlyHint=True),
)
async def gh_whoami() -> dict:
ctx = get_context()
resp = await ctx.dispatch("github", HttpRequest(
method=HttpMethod.GET,
path="/user"
))
if resp.success:
u = resp.response.body
return {"login": u.get("login"), "name": u.get("name")}
return {"error": resp.error.message if resp.error else "Request failed"}
@tool(
description="List repositories for the authenticated user",
tags=["repos", "read"],
annotations=ToolAnnotations(readOnlyHint=True),
)
async def gh_list_repos(per_page: int = 10) -> list[dict]:
ctx = get_context()
resp = await ctx.dispatch("github", HttpRequest(
method=HttpMethod.GET,
path=f"/user/repos?per_page={per_page}&sort=updated"
))
if resp.success and isinstance(resp.response.body, list):
return [
{"name": x.get("name"), "stars": x.get("stargazers_count", 0)}
for x in resp.response.body
]
return []
X (Twitter) API
Read-only X API tools using OAuth 2.0 App-Only authentication.Environment
Report incorrect code
Copy
Ask AI
# .env
X_BEARER_TOKEN=AAAA... # App-only bearer token
X_API_KEY=... # Optional: for user-context endpoints
X_API_KEY_SECRET=... # Optional: for user-context endpoints
Free tier limits: 100 tweet reads, 500 writes/month. For production, consider Basic ($100/mo) or Pro tier.
Connection
Report incorrect code
Copy
Ask AI
# x.py
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"
Tools
Report incorrect code
Copy
Ask AI
@tool(
description="Get an X user by their username",
tags=["user", "read"],
annotations=ToolAnnotations(readOnlyHint=True),
)
async def x_get_user_by_username(username: str) -> dict:
ctx = get_context()
resp = await ctx.dispatch("x", HttpRequest(
method=HttpMethod.GET,
path=f"/2/users/by/username/{username}?user.fields={DEFAULT_USER_FIELDS}"
))
if resp.success:
return {"data": resp.response.body.get("data")}
return {"error": resp.error.message if resp.error else "Request failed"}
@tool(
description="Search recent tweets (last 7 days)",
tags=["search", "read"],
annotations=ToolAnnotations(readOnlyHint=True),
)
async def x_search_recent(query: str, max_results: int = 10) -> dict:
from urllib.parse import quote
ctx = get_context()
max_results = max(10, min(100, max_results))
resp = await ctx.dispatch("x", HttpRequest(
method=HttpMethod.GET,
path=f"/2/tweets/search/recent?query={quote(query)}&tweet.fields={DEFAULT_TWEET_FIELDS}&max_results={max_results}"
))
if resp.success:
body = resp.response.body or {}
return {"data": body.get("data"), "meta": body.get("meta")}
return {"error": resp.error.message if resp.error else "Request failed"}
@tool(
description="Get a user's recent tweets",
tags=["tweet", "read"],
annotations=ToolAnnotations(readOnlyHint=True),
)
async def x_get_user_tweets(user_id: str, max_results: int = 10) -> dict:
ctx = get_context()
max_results = max(5, min(100, max_results))
resp = await ctx.dispatch("x", HttpRequest(
method=HttpMethod.GET,
path=f"/2/users/{user_id}/tweets?tweet.fields={DEFAULT_TWEET_FIELDS}&max_results={max_results}"
))
if resp.success:
body = resp.response.body or {}
return {"data": body.get("data"), "meta": body.get("meta")}
return {"error": resp.error.message if resp.error else "Request failed"}
Gmail (OAuth 2.0)
Gmail MCP server with true OAuth 2.0 user authentication.Environment
Report incorrect code
Copy
Ask AI
# .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
Setup
- Enable Gmail API: Go to Gmail API and click “Enable”
- Create OAuth credentials: Go to APIs & Services → Credentials and:
- Click “Create Credentials” → “OAuth client ID”
- Application type: “Web application”
- Add authorized redirect URIs for your deployment
- Copy the Client ID and Client Secret to your
.env
- Configure consent screen: Set up the OAuth consent screen with the Gmail scopes listed above
Server
Report incorrect code
Copy
Ask 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)
Connection
Report incorrect code
Copy
Ask AI
# gmail.py
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}",
)
Tools
Report incorrect code
Copy
Ask AI
@tool(
description="List recent emails",
tags=["email", "read"],
annotations=ToolAnnotations(readOnlyHint=True),
)
async def gmail_list_messages(max_results: int = 10, query: str = "") -> dict:
ctx = get_context()
path = f"/gmail/v1/users/me/messages?maxResults={max_results}"
if query:
path += f"&q={query}"
resp = await ctx.dispatch("gmail", HttpRequest(
method=HttpMethod.GET,
path=path
))
if resp.success:
return {"messages": resp.response.body.get("messages", [])}
return {"error": resp.error.message if resp.error else "Request failed"}
@tool(
description="Get email details by ID",
tags=["email", "read"],
annotations=ToolAnnotations(readOnlyHint=True),
)
async def gmail_get_message(message_id: str) -> dict:
ctx = get_context()
resp = await ctx.dispatch("gmail", HttpRequest(
method=HttpMethod.GET,
path=f"/gmail/v1/users/me/messages/{message_id}"
))
if resp.success:
return {"message": resp.response.body}
return {"error": resp.error.message if resp.error else "Request failed"}
Running the examples
All examples follow the same pattern:Report incorrect code
Copy
Ask AI
# Clone the repo
git clone https://github.com/dedalus-labs/example-dedalus-mcp
cd example-dedalus-mcp
# For the main example (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
http://127.0.0.1:8080/mcp.