将现有的LlamaIndex工作流和工具转换为MCP

将您的LlamaIndex工具和工作流转换为MCP服务器，以实现更广泛的生态系统兼容性。

转换工作流

使用 workflow_as_mcp 将任意 LlamaIndex 工作流转换为 FastMCP 服务器：

from workflows import Context, Workflow, step
from workflows.events import StartEvent, StopEvent
from llama_index.tools.mcp.utils import workflow_as_mcp


class QueryEvent(StartEvent):
    query: str


class SimpleWorkflow(Workflow):
    @step
    def process_query(self, ctx: Context, ev: QueryEvent) -> StopEvent:
        result = f"Processed: {ev.query}"
        return StopEvent(result=result)


# Convert to MCP server
workflow = SimpleWorkflow()
mcp = workflow_as_mcp(workflow, start_event_model=QueryEvent)

如果您直接使用 FastMCP，效果大致如下：

from fastmcp import FastMCP

# Workflow definition
...

mcp = FastMCP("Demo 🚀")
workflow = SimpleWorkflow()


@mcp.tool
async def run_my_workflow(input_args: QueryEvent) -> str:
    """Add two numbers"""
    if isintance(input_args, dict):
        input_args = QueryEvent.model_validate(input_args)
    result = await workflow.run(start_event=input_args)
    return str(result)


if __name__ == "__main__":
    mcp.run()

转换独立工具

我们也可以使用 FastMCP 直接将现有函数和工具转换为 MCP 端点输入：

from fastmcp import FastMCP
from llama_index.tools.notion import NotionToolSpec

# Get tools from ToolSpec
tool_spec = NotionToolSpec(integration_token="your_token")
tools = tool_spec.to_tool_list()

# Create MCP server
mcp_server = FastMCP("Tool Server")

# Register tools
for tool in tools:
    mcp_server.tool(
        name=tool.metadata.name, description=tool.metadata.description
    )(tool.real_fn)

运行MCP服务器

您可以通过命令行界面启动服务器（这对调试也非常有用！）：

# Install MCP CLI
pip install "mcp[cli]"

# Run server
mcp run your-server.py