Dynamic Capability Loading

Static tool lists don't scale. Your agent needs different capabilities for "analyze this CSV" versus "deploy to production." Dynamic loading gives agents the right tools at the right time.

The Loading Strategy

class DynamicSkillLoader:
    def __init__(self, skills_dir: str = "skills"):
        self.skills_dir = Path(skills_dir)
        self.skill_registry = {}
        self.loaded_tools = {}
        self.scan_skills()

    def scan_skills(self):
        """Build registry of available skills."""
        for skill_path in self.skills_dir.glob("*/SKILL.md"):
            skill_name = skill_path.parent.name
            self.skill_registry[skill_name] = {
                "path": skill_path.parent,
                "manifest": self.parse_manifest(skill_path),
                "loaded": False
            }

    def load_skill(self, skill_name: str) -> list[dict]:
        """Dynamically import and initialize a skill."""
        if skill_name not in self.skill_registry:
            raise ValueError(f"Unknown skill: {skill_name}")

        skill = self.skill_registry[skill_name]
        if skill["loaded"]:
            return self.loaded_tools[skill_name]

        # Import the skill module
        module_path = skill["path"] / "tools.py"
        spec = importlib.util.spec_from_file_location(skill_name, module_path)
        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)

        # Extract tool definitions
        tools = module.get_tools()
        self.loaded_tools[skill_name] = tools
        skill["loaded"] = True

        return tools

Intent-Based Tool Discovery

Let the model decide what it needs:

async def discover_tools_for_intent(user_message: str, loader: DynamicSkillLoader) -> list[dict]:
    """Use a small model to determine required skills."""

    # Build skill summary for the classifier
    skill_summaries = "\n".join([
        f"- {name}: {info['manifest']['description']}"
        for name, info in loader.skill_registry.items()
    ])

    response = await llm.chat(
        model="claude-3-haiku-20240307",  # Fast, cheap classifier
        messages=[{
            "role": "user",
            "content": f"""Given this user request, which skills are needed?

User request: {user_message}

Available skills:
{skill_summaries}

Return a JSON array of skill names. Only include skills directly relevant to the task.
Example: ["filesystem", "web"]"""
        }]
    )

    needed_skills = json.loads(response.content)

    # Load only the required skills
    tools = []
    for skill_name in needed_skills:
        tools.extend(loader.load_skill(skill_name))

    return tools

Lazy Loading with Caching

class CachedSkillLoader:
    def __init__(self):
        self.loader = DynamicSkillLoader()
        self.tool_cache = {}
        self.cache_ttl = 300  # 5 minutes

    async def get_tools(self, user_message: str) -> list[dict]:
        """Get tools with caching for repeated patterns."""
        cache_key = self.compute_intent_hash(user_message)

        if cache_key in self.tool_cache:
            entry = self.tool_cache[cache_key]
            if time.time() - entry["timestamp"] < self.cache_ttl:
                return entry["tools"]

        tools = await discover_tools_for_intent(user_message, self.loader)
        self.tool_cache[cache_key] = {
            "tools": tools,
            "timestamp": time.time()
        }

        return tools

    def compute_intent_hash(self, message: str) -> str:
        """Hash based on key intent words, not exact message."""
        # Extract intent keywords for cache matching
        keywords = set(re.findall(r'\b(file|web|code|email|database)\b', message.lower()))
        return hashlib.md5(",".join(sorted(keywords)).encode()).hexdigest()

MCP Server Hot-Loading

Connect to MCP servers on demand:

class MCPDynamicConnector:
    def __init__(self, config_path: str):
        self.config = json.load(open(config_path))
        self.active_connections = {}

    async def connect_server(self, server_name: str):
        """Start MCP server connection on demand."""
        if server_name in self.active_connections:
            return self.active_connections[server_name]

        server_config = self.config["mcpServers"][server_name]

        # Start the server process
        process = await asyncio.create_subprocess_exec(
            server_config["command"],
            *server_config.get("args", []),
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE
        )

        # Initialize MCP connection
        connection = MCPConnection(process)
        await connection.initialize()

        self.active_connections[server_name] = connection
        return connection

    async def get_tools_from_server(self, server_name: str) -> list[dict]:
        """Fetch tools from a specific MCP server."""
        connection = await self.connect_server(server_name)
        return await connection.list_tools()

    async def disconnect_idle_servers(self, idle_threshold: int = 300):
        """Clean up servers not used recently."""
        now = time.time()
        for name, conn in list(self.active_connections.items()):
            if now - conn.last_used > idle_threshold:
                await conn.close()
                del self.active_connections[name]

The Full Pipeline

async def process_with_dynamic_tools(user_message: str):
    """Complete flow: discover → load → execute."""

    # 1. Classify intent and discover needed tools
    tools = await cached_loader.get_tools(user_message)

    # 2. Check if MCP servers are needed
    if needs_external_capability(user_message):
        mcp_tools = await mcp_connector.get_tools_from_server("relevant-server")
        tools.extend(mcp_tools)

    # 3. Call LLM with dynamically assembled toolset
    response = await llm.chat(
        messages=[{"role": "user", "content": user_message}],
        tools=tools
    )

    # 4. Execute tool calls
    while response.tool_calls:
        results = await execute_tools(response.tool_calls)
        response = await llm.chat(
            messages=[...],  # Include tool results
            tools=tools
        )

    # 5. Cleanup idle connections
    await mcp_connector.disconnect_idle_servers()

    return response.content

Nerd Note: This is how Claude Code works. It doesn't load all 50+ tools at once—it activates what's needed based on your request.

Next module: Taking agents to production. :::