MCPStoreContext 类¶

class MCPStoreContext:
    """
    MCPStore 操作上下文
    提供服务管理、工具操作、LangChain集成等功能
    """
    def __init__(self, store: MCPStore, context_type: ContextType = ContextType.STORE, agent_id: str = None):
        """
        初始化上下文

        Args:
            store: MCPStore 实例
            context_type: 上下文类型 (STORE 或 AGENT)
            agent_id: Agent ID (仅在 AGENT 模式下使用)
        """

from enum import Enum

class ContextType(Enum):
    STORE = "store"    # Store 级别上下文
    AGENT = "agent"    # Agent 级别上下文

def add_service(self, config: Union[ServiceConfigUnion, List[str], None] = None,
               json_file: str = None) -> 'MCPStoreContext'

def list_services() -> List[ServiceInfo]

def get_service_info(name: str) -> Any

def restart_service(name: str) -> bool

def check_services() -> Dict[str, Any]

def delete_service(name: str) -> bool

def update_service(name: str, config: Dict[str, Any]) -> bool

def patch_service(name: str, updates: Dict[str, Any]) -> bool

def get_service_status(name: str) -> dict

def show_config(scope: str = "all") -> Dict[str, Any]

def reset_config(scope: str = "all") -> bool

def show_mcpconfig() -> Dict[str, Any]

def list_tools() -> List[ToolInfo]

def call_tool(tool_name: str, args: Union[Dict[str, Any], str] = None, **kwargs) -> Any

def use_tool(tool_name: str, args: Union[Dict[str, Any], str] = None, **kwargs) -> Any

def get_tools_with_stats() -> Dict[str, Any]

def batch_add_services(services: List[Union[str, Dict[str, Any]]]) -> Dict[str, Any]

def get_system_stats() -> Dict[str, Any]

def for_langchain() -> 'LangChainAdapter'

# 获取 LangChain 工具
tools = store.for_store().for_langchain().list_tools()

# 链式调用
tools = (store.for_store()
         .add_service(config)
         .for_langchain()
         .list_tools())

# 异步服务操作
async def add_service_async(config, json_file=None) -> 'MCPStoreContext'
async def list_services_async() -> List[ServiceInfo]
async def get_service_info_async(name: str) -> Any
async def restart_service_async(name: str) -> bool
async def check_services_async() -> Dict[str, Any]

# 异步工具操作
async def list_tools_async() -> List[ToolInfo]
async def call_tool_async(tool_name: str, args=None, **kwargs) -> Any
async def use_tool_async(tool_name: str, args=None, **kwargs) -> Any
async def get_tools_with_stats_async() -> Dict[str, Any]
async def batch_add_services_async(services) -> Dict[str, Any]

from mcpstore import MCPStore

store = MCPStore.setup_store()

# 获取 Store 上下文
store_context = store.for_store()

# 添加服务
store_context.add_service({
    "name": "weather-api",
    "url": "https://weather.example.com/mcp"
})

# 列出服务和工具
services = store_context.list_services()
tools = store_context.list_tools()

# 调用工具
result = store_context.call_tool("get_weather", {"city": "北京"})

print(f"Store 级别: {len(services)} 服务, {len(tools)} 工具")

from mcpstore import MCPStore

store = MCPStore.setup_store()

# 获取 Agent 上下文
agent_context = store.for_agent("my_agent")

# Agent 级别操作
agent_context.add_service({
    "name": "agent-tool",
    "url": "https://agent.example.com/mcp"
})

# Agent 只能看到自己的服务
agent_services = agent_context.list_services()
agent_tools = agent_context.list_tools()

print(f"Agent 级别: {len(agent_services)} 服务, {len(agent_tools)} 工具")

from mcpstore import MCPStore

store = MCPStore.setup_store()

# Store 级别链式调用
result = (store.for_store()
          .add_service({"name": "service1", "url": "https://api1.com/mcp"})
          .add_service({"name": "service2", "url": "https://api2.com/mcp"})
          .list_tools())

print(f"链式调用获得 {len(result)} 个工具")

# Agent 级别链式调用
agent_result = (store.for_agent("test_agent")
                .add_service({"name": "agent_service", "url": "https://agent.com/mcp"})
                .call_tool("some_tool", {"param": "value"}))

print(f"Agent 工具调用结果: {agent_result}")

import asyncio
from mcpstore import MCPStore

async def async_operations():
    store = MCPStore.setup_store()

    # 异步添加服务
    context = await store.for_store().add_service_async({
        "name": "async-service",
        "url": "https://async.example.com/mcp"
    })

    # 异步获取工具列表
    tools = await context.list_tools_async()

    # 异步调用工具
    result = await context.call_tool_async("async_tool", {"data": "test"})

    print(f"异步操作: {len(tools)} 工具, 结果: {result}")

# 运行异步示例
asyncio.run(async_operations())