Registering Tools and Lead Sources

MCP tools

Register tools in your pack.py via context.mcp_client.register_builtin_handler():

def register(self, context: PackContext) -> None:
    if context.mcp_client is None:
        return

    async def weather_forecast(city: str, days: int = 3) -> str:
        import json
        # Your implementation here
        return json.dumps({"city": city, "forecast": "sunny", "days": days})

    context.mcp_client.register_builtin_handler(
        "weather_forecast",
        weather_forecast,
        description="Get weather forecast for a city",
        input_schema={
            "type": "object",
            "properties": {
                "city": {"type": "string", "description": "City name"},
                "days": {"type": "integer", "description": "Forecast days (1-7)", "default": 3},
            },
            "required": ["city"],
        },
        risk_level="green",
    )

Handler requirements

• Must be async def
• Return a JSON-serializable string
• Parameters must match the input_schema properties
• Handle errors gracefully — return error info as JSON, do not crash

Risk levels

The Gatekeeper classifies every tool call. Set risk_level to control the default:

If you omit risk_level, unknown tools default to orange (user approval required).

Lead sources

Implement a LeadSource subclass to add a new channel to the lead hunting system.

Required class variables

from cognithor.leads.source import LeadSource
from cognithor.leads.models import Lead

class MySource(LeadSource):
    source_id = "my-platform"
    display_name = "My Platform"
    icon = "forum"           # Material icon name
    color = "#FF6B35"        # Hex color for UI
    capabilities = frozenset({"scan"})

Required method: scan()

    async def scan(
        self,
        *,
        config: dict,
        product: str,
        product_description: str,
        min_score: int,
    ) -> list[Lead]:
        posts = await self._fetch_posts(config)
        leads = []
        for post in posts:
            score = await self._score(post, product, product_description)
            if score >= min_score:
                leads.append(Lead(
                    post_id=f"my-platform-{post['id']}",
                    source_id=self.source_id,
                    title=post["title"],
                    url=post["url"],
                    intent_score=score,
                    body=post.get("body", ""),
                    author=post.get("author", ""),
                ))
        return leads

Optional methods

Add capabilities to enable these:

Registration

def register(self, context: PackContext) -> None:
    if context.leads:
        self._source = MySource()
        context.leads.register_source(self._source)

def unregister(self, context: PackContext) -> None:
    if context.leads and self._source:
        context.leads.unregister_source(self._source.source_id)