​
FastMCP Provider

FastMCP is a provider that enables direct tool execution through the Model Context Protocol (MCP). Unlike ADK which generates workflows from natural language, FastMCP focuses on executing specific tools directly.

​
When to Use FastMCP

Use FastMCP when you:

  • Have MCP-compatible tools you want to execute
  • Need direct tool execution without AI workflow generation
  • Want to leverage the standardized MCP protocol
  • Are building integrations with existing MCP servers

​
Quick Start

​
1. Install the SDK

pip install kubiya-workflow-sdk

​
2. Start MCP Server

# Point to your MCP server implementation
export KUBIYA_MCP_SERVER_PATH="/path/to/mcp/server"

# Start the FastMCP provider
python -m kubiya_workflow_sdk.providers.fastmcp.provider

​
3. Execute Tools

from kubiya_workflow_sdk.providers import get_provider

# Initialize FastMCP provider
mcp = get_provider("fastmcp", server_path="/path/to/mcp/server")

# Execute a tool directly
result = await mcp.execute_tool(
    tool_name="file_operations",
    arguments={
        "action": "read",
        "path": "/data/report.csv"
    }
)

print(result)  # Tool execution result

​
Available MCP Tools

FastMCP can work with any MCP-compatible tools:

File System

Read, write, and manage files

await mcp.execute_tool("fs_read", {"path": "/data/file.txt"})

GitHub

Interact with GitHub repositories

await mcp.execute_tool("github_create_issue", {
    "repo": "owner/repo",
    "title": "Bug report"
})

Database

Query and update databases

await mcp.execute_tool("db_query", {
    "query": "SELECT * FROM users"
})

Custom Tools

Your own MCP-compatible tools

await mcp.execute_tool("custom_tool", {
    "param": "value"
})

​
Integration with Workflows

You can use FastMCP within Kubiya workflows:

from kubiya_workflow_sdk.dsl import workflow, step
from kubiya_workflow_sdk.providers import get_provider

@workflow
def data_pipeline():
    # Step 1: Read data using MCP tool
    read_data = step("read_csv").mcp_tool(
        tool="fs_read",
        arguments={"path": "/data/input.csv"}
    )
    
    # Step 2: Process data (your logic)
    process = step("process").python(
        code="""
        import pandas as pd
        df = pd.read_csv(io.StringIO(inputs['csv_data']))
        result = df.groupby('category').sum()
        print(result.to_json())
        """,
        inputs={"csv_data": "${read_data.content}"}
    ).depends("read_csv")
    
    # Step 3: Write results using MCP tool
    write_result = step("write_results").mcp_tool(
        tool="fs_write",
        arguments={
            "path": "/data/output.json",
            "content": "${process.stdout}"
        }
    ).depends("process")

​
FastMCP vs ADK

FeatureFastMCPADK
PurposeDirect tool executionAI workflow generation
InputTool name + parametersNatural language
Use Case”Execute database query""Analyze sales data and create report”
AI RequiredNoYes
StreamingTool output onlyFull workflow progress

​
Configuration

​
Environment Variables

# MCP server location
export KUBIYA_MCP_SERVER_PATH="/usr/local/bin/mcp-server"

# Server configuration
export KUBIYA_FASTMCP_PORT=8002
export KUBIYA_FASTMCP_HOST="0.0.0.0"

# Tool discovery
export KUBIYA_MCP_AUTO_DISCOVER=true

​
Custom MCP Server

Create your own MCP-compatible tool:

# my_mcp_tool.py
from kubiya_workflow_sdk.mcp import MCPTool, MCPServer

class MyCustomTool(MCPTool):
    name = "my_tool"
    description = "Custom tool for specific task"
    
    async def execute(self, **kwargs):
        # Your tool logic here
        return {"result": "success", "data": kwargs}

# Register and run
server = MCPServer()
server.register_tool(MyCustomTool())
server.run()

​
Next Steps