import asyncio
import os
from dedalus_labs import AsyncDedalus, DedalusRunner
from dedalus_labs.utils.stream import stream_async
from dedalus_mcp.auth import Connection, SecretKeys, SecretValues
from dotenv import load_dotenv

load_dotenv()

# The user we are researching
# !! Remember to update this to a real account !!
name = "Jane Doe"
x_username = "@JaneDoe"

# Connection: schema for X/Twitter API
# Note: Some providers may also require an additional "auth_header_format" parameter in Connection(). X does not.
x = Connection(
    name="x",
    secrets=SecretKeys(token="X_BEARER_TOKEN"),
    base_url="https://api.x.com",
)

# 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 main():
    client = AsyncDedalus(timeout=900) # 15 minutes (unit = seconds)
    runner = DedalusRunner(client)

    response = runner.run(
        input=f"""You are an expert CRM Research Agent. Your objective is to identify high-value second-degree connections for {name} ({x_username}).

        1. Use x-api-mcp (e.g., x_get_followers, x_get_user_by_username) to go through all of {name}'s followers.
        2. Research each follower's own network (second-degree) and their professional background.
        3. If X info is sparse, use brave-search-mcp to find more details on the web. Explicitly mention it when you cannot access X.
        4. Find at least 15 high-value connections for {name} ({x_username}). These high-value connections should be users that are not already in {name}'s network.
        5. For EVERY high-value connection found, format the output in a bulleted summary:
           - Name: [Full name]
           - X/Social: [Handle/Link]
           - Bio: [Brief summary]
           - Key Interests: [Topic 1, Topic 2]
           - Connection Logic: [Why they are a good lead]""",

        model="anthropic/claude-opus-4-5",
        mcp_servers=["windsor/x-api-mcp", "tsion/brave-search-mcp"],
        credentials=[x_secrets],
        stream=True,
    )

    await stream_async(response)

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

...

response = runner.run(
        input=f"""You are an expert CRM Research Agent. Your objective is to identify high-value second-degree connections for {name} ({x_username}).

        1. Use x-api-mcp (e.g., x_get_followers, x_get_user_by_username) to go through all of {name}'s followers.
        2. Research each follower's own network (second-degree) and their professional background.
        3. If X info is sparse, use brave-search-mcp to find more details on the web. Explicitly mention it when you cannot access X.
        4. Find at least 15 high-value connections for {name} ({x_username}). These high-value connections should be users that are not already in {name}'s network.
        5. For EVERY high-value connection found, format the output in a bulleted summary:
           - Name: [Full name]
           - X/Social: [Handle/Link]
           - Bio: [Brief summary]
           - Key Interests: [Topic 1, Topic 2]
           - Connection Logic: [Why they are a good lead]""",

        model="anthropic/claude-opus-4-5",
        mcp_servers=["windsor/x-api-mcp", "tsion/brave-search-mcp"],
        credentials=[x_secrets],
        stream=True,
    )
...

@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"}

# Connection: schema for X/Twitter API
# Note: Some providers may also require an additional "auth_header_format" parameter in Connection(). X does not.
x = Connection(
    name="x",
    secrets=SecretKeys(token="X_BEARER_TOKEN"),
    base_url="https://api.x.com",
)

# 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", ""))

...
		model="anthropic/claude-opus-4-5",
    mcp_servers=["tsion/brave-search-mcp", "windsor/x-api-mcp"],
    credentials=[x_secrets],
...

pip install dedalus-labs dedalus-mcp python-dotenv

X_BEARER_TOKEN=your-x-bearer-token

DEDALUS_API_KEY=your-api-key
DEDALUS_AS_URL=https://as.dedaluslabs.ai

import asyncio
import os
from dedalus_labs import AsyncDedalus, DedalusRunner
from dedalus_labs.utils.stream import stream_async
from dedalus_mcp.auth import Connection, SecretKeys, SecretValues
from dotenv import load_dotenv

load_dotenv()

# The user we are researching
# !! Remember to update this to a real account !!
name = "Jane Doe"
x_username = "@JaneDoe"

# Connection: schema for X/Twitter API
# Note: Some providers may also require an additional "auth_header_format" parameter in Connection(). X does not.
x = Connection(
    name="x",
    secrets=SecretKeys(token="X_BEARER_TOKEN"),
    base_url="https://api.x.com",
)

# 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 main():
    client = AsyncDedalus(timeout=900) # 15 minutes (unit = seconds)
    runner = DedalusRunner(client)

    response = runner.run(
        input=f"""You are an expert CRM Research Agent. Your objective is to identify high-value second-degree connections for {name} ({x_username}).

        1. Use x-api-mcp (e.g., x_get_followers, x_get_user_by_username) to go through all of {name}'s followers.
        2. Research each follower's own network (second-degree) and their professional background.
        3. If X info is sparse, use brave-search-mcp to find more details on the web. Explicitly mention it when you cannot access X.
        4. Find at least 15 high-value connections for {name} ({x_username}). These high-value connections should be users that are not already in {name}'s network.
        5. For EVERY high-value connection found, format the output in a bulleted summary:
           - Name: [Full name]
           - X/Social: [Handle/Link]
           - Bio: [Brief summary]
           - Key Interests: [Topic 1, Topic 2]
           - Connection Logic: [Why they are a good lead]""",

        model="anthropic/claude-opus-4-5",
        mcp_servers=["windsor/x-api-mcp", "tsion/brave-search-mcp"],
        credentials=[x_secrets],
        stream=True,
    )

    await stream_async(response)

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

...

response = runner.run(
        input=f"""You are an expert CRM Research Agent. Your objective is to identify high-value second-degree connections for {name} ({x_username}).

        1. Use x-api-mcp (e.g., x_get_followers, x_get_user_by_username) to go through all of {name}'s followers.
        2. Research each follower's own network (second-degree) and their professional background.
        3. If X info is sparse, use brave-search-mcp to find more details on the web. Explicitly mention it when you cannot access X.
        4. Find at least 15 high-value connections for {name} ({x_username}). These high-value connections should be users that are not already in {name}'s network.
        5. For EVERY high-value connection found, format the output in a bulleted summary:
           - Name: [Full name]
           - X/Social: [Handle/Link]
           - Bio: [Brief summary]
           - Key Interests: [Topic 1, Topic 2]
           - Connection Logic: [Why they are a good lead]""",

        model="anthropic/claude-opus-4-5",
        mcp_servers=["windsor/x-api-mcp", "tsion/brave-search-mcp"],
        credentials=[x_secrets],
        stream=True,
    )
...

@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"}

# Connection: schema for X/Twitter API
# Note: Some providers may also require an additional "auth_header_format" parameter in Connection(). X does not.
x = Connection(
    name="x",
    secrets=SecretKeys(token="X_BEARER_TOKEN"),
    base_url="https://api.x.com",
)

# 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", ""))

...
		model="anthropic/claude-opus-4-5",
    mcp_servers=["tsion/brave-search-mcp", "windsor/x-api-mcp"],
    credentials=[x_secrets],
...

pip install dedalus-labs dedalus-mcp python-dotenv

X_BEARER_TOKEN=your-x-bearer-token

DEDALUS_API_KEY=your-api-key
DEDALUS_AS_URL=https://as.dedaluslabs.ai