stainless-SDK

Mcp服务器Python API库

)](https://pypi.org/project/mcp_servers/)

Mcp服务器Python库提供了从任何Python 3.8到Mcp服务器REST API的方便访问+ 应用程序。该库包括所有请求参数和响应字段的类型定义， 并提供同步和异步客户端，由 httpx.

它是通过以下方式生成的 不锈钢.

文档

此库的完整API可在中找到 api.md.

安装

# install from this staging repo
pip install git+ssh://git@github.com/stainless-sdks/mcp-servers-python.git

\[！注意\] 一旦这个包裹 发布到PyPI，这将变成： pip install --pre mcp_servers

用法

此库的完整API可在中找到 api.md.

import os
from mcp_servers import McpServers

client = McpServers(
    api_key=os.environ.get("PETSTORE_API_KEY"),  # This is the default and can be omitted
)

order = client.store.order.create(
    pet_id=1,
    quantity=1,
    status="placed",
)
print(order.id)

虽然你可以提供 api_key 关键字参数， 我们建议使用 python dotenv 增加 PETSTORE_API_KEY="My API Key" 到你的 .env 文件 以便您的API密钥不会存储在源代码管理中。

异步用法

只需导入 AsyncMcpServers 而不是 McpServers 和使用 await 对于每个API调用：

import os
import asyncio
from mcp_servers import AsyncMcpServers

client = AsyncMcpServers(
    api_key=os.environ.get("PETSTORE_API_KEY"),  # This is the default and can be omitted
)

async def main() -> None:
    order = await client.store.order.create(
        pet_id=1,
        quantity=1,
        status="placed",
    )
    print(order.id)

asyncio.run(main())

同步和异步客户端之间的功能在其他方面是相同的。

通过aiohttp

默认情况下，异步客户端使用 httpx 对于HTTP请求。但是，为了提高并发性能，您还可以使用 aiohttp 作为HTTP后端。

您可以通过安装来启用此功能 aiohttp:

# install from this staging repo
pip install 'mcp_servers[aiohttp] @ git+ssh://git@github.com/stainless-sdks/mcp-servers-python.git'

然后，您可以通过实例化客户端来启用它 http_client=DefaultAioHttpClient():

import asyncio
from mcp_servers import DefaultAioHttpClient
from mcp_servers import AsyncMcpServers

async def main() -> None:
    async with AsyncMcpServers(
        api_key="My API Key",
        http_client=DefaultAioHttpClient(),
    ) as client:
        order = await client.store.order.create(
            pet_id=1,
            quantity=1,
            status="placed",
        )
        print(order.id)

asyncio.run(main())

使用类型

嵌套请求参数为 类型图片.答复如下 Pydantic模型 它还为以下内容提供了辅助方法：

• 序列化回JSON， model.to_json()
• 转换为词典， model.to_dict()

键入的请求和响应在编辑器中提供自动补全和文档。如果你想在VS Code中看到类型错误，以帮助更早地发现错误，请设置 python.analysis.typeCheckingMode 向 basic.

嵌套参数

嵌套参数是字典，使用 TypedDict例如：

from mcp_servers import McpServers

client = McpServers()

pet = client.pet.create(
    name="doggie",
    photo_urls=["string"],
    category={},
)
print(pet.category)

处理错误

当库无法连接到API时（例如，由于网络连接问题或超时） mcp_servers.APIConnectionError 提高。

当API返回非成功状态代码（即4xx或5xx response），一个子类 mcp_servers.APIStatusError 上升，包含 status_code 和 response 物业。

所有错误都继承自 mcp_servers.APIError.

import mcp_servers
from mcp_servers import McpServers

client = McpServers()

try:
    client.store.list_inventory()
except mcp_servers.APIConnectionError as e:
    print("The server could not be reached")
    print(e.__cause__)  # an underlying Exception, likely raised within httpx.
except mcp_servers.RateLimitError as e:
    print("A 429 status code was received; we should back off a bit.")
except mcp_servers.APIStatusError as e:
    print("Another non-200-range status code was received")
    print(e.status_code)
    print(e.response)

错误代码如下：

重试

默认情况下，某些错误会自动重试2次，并具有短暂的指数回退。 连接错误（例如，由于网络连接问题），408请求超时，409冲突， 默认情况下，429速率限制和>=500内部错误都会重试。

您可以使用 max_retries 配置或禁用重试设置的选项：

from mcp_servers import McpServers

# Configure the default for all requests:
client = McpServers(
    # default is 2
    max_retries=0,
)

# Or, configure per-request:
client.with_options(max_retries=5).store.list_inventory()

超时

默认情况下，请求在1分钟后超时。您可以使用 timeout 选项， 它接受浮点数或 httpx.Timeout 对象：

from mcp_servers import McpServers

# Configure the default for all requests:
client = McpServers(
    # 20 seconds (default is 1 minute)
    timeout=20.0,
)

# More granular control:
client = McpServers(
    timeout=httpx.Timeout(60.0, read=5.0, write=10.0, connect=2.0),
)

# Override per-request:
client.with_options(timeout=5.0).store.list_inventory()

在超时时 APITimeoutError 被抛出。

请注意，超时请求是 默认情况下重试两次.

高级

日志记录

我们使用标准库 logging 模块。

您可以通过设置环境变量来启用日志记录 MCP_SERVERS_LOG 向 info.

$ export MCP_SERVERS_LOG=info

或者 debug 以获得更详细的日志记录。

如何判断是否 None 手段 null 或失踪

在API响应中，字段可以显式地 null，或完全缺失；无论哪种情况，它的值都是 None 在这个图书馆。你可以用以下方式区分这两种情况 .model_fields_set:

if response.my_field is None:
  if 'my_field' not in response.model_fields_set:
    print('Got json like {}, without a "my_field" key present at all.')
  else:
    print('Got json like {"my_field": null}.')

访问原始响应数据（例如标头）

可以通过前缀访问“原始”响应对象 .with_raw_response. 对于任何HTTP方法调用。，

from mcp_servers import McpServers

client = McpServers()
response = client.store.with_raw_response.list_inventory()
print(response.headers.get('X-My-Header'))

store = response.parse()  # get the object that `store.list_inventory()` would have returned
print(store)

这些方法返回一个 APIResponse 对象。

异步客户端返回一个 AsyncAPIResponse 结构相同，唯一的区别是 await读取响应内容的可行方法。

.with_streaming_response

当您发出请求时，上述接口会急切地读取完整的响应正文，这可能并不总是您想要的。

要流式传输响应正文，请使用 .with_streaming_response 相反，这需要一个上下文管理器，并且只在您调用时读取响应正文 .read(), .text(), .json(), .iter_bytes(), .iter_text(), .iter_lines() 或 .parse()。在异步客户端中，这些是异步方法。

with client.store.with_streaming_response.list_inventory() as response:
    print(response.headers.get("X-My-Header"))

    for line in response.iter_lines():
        print(line)

需要上下文管理器，以便可靠地关闭响应。

提出定制/未记录的请求

键入此库是为了方便访问文档化的API。

如果需要访问未记录的端点、参数或响应属性，仍然可以使用该库。

未记录的端点

要向未记录的端点发出请求，您可以使用 client.get, client.post，及其他 http动词。发出此请求时，将尊重客户端上的选项（如重试）。

import httpx

response = client.post(
    "/foo",
    cast_to=httpx.Response,
    body={"my_param": True},
)

print(response.headers.get("x-foo"))

未记录的请求参数

如果你想显式地发送一个额外的参数，你可以用 extra_query, extra_body，以及 extra_headers 请求 选项。

未记录的响应属性

要访问未记录的响应属性，您可以访问以下额外字段 response.unknown_prop.你 还可以使用以下命令将Pydantic模型上的所有额外字段作为字典获取 response.model_extra.

配置HTTP客户端

您可以直接覆盖 httpx客户端 为您的用例进行自定义，包括：

• 对……的支持 代理
• 自定义 运输
• 额外 先进的 功能

import httpx
from mcp_servers import McpServers, DefaultHttpxClient

client = McpServers(
    # Or use the `MCP_SERVERS_BASE_URL` env var
    base_url="http://my.test.server.example.com:8083",
    http_client=DefaultHttpxClient(
        proxy="http://my.test.proxy.example.com",
        transport=httpx.HTTPTransport(local_address="0.0.0.0"),
    ),
)

您还可以使用以下命令根据每个请求自定义客户端 with_options():

client.with_options(http_client=DefaultHttpxClient(...))

管理HTTP资源

默认情况下，每当客户端发生以下情况时，库都会关闭底层HTTP连接 垃圾收集。您可以使用手动关闭客户端 .close() 如果需要，可以使用方法，或者使用退出时关闭的上下文管理器。

from mcp_servers import McpServers

with McpServers() as client:
  # make requests here
  ...

# HTTP client is now closed

版本控制

此套餐通常遵循 学期 尽管某些向后不兼容的更改可能会作为次要版本发布：

1. 仅影响静态类型而不破坏运行时行为的更改。
2. 对库内部的更改，这些更改在技术上是公开的，但不是为外部使用而设计或记录的。 _（请在GitHub上发布一个问题，让我们知道您是否依赖这些内部机制。）_
3. 我们预计在实践中不会影响绝大多数用户的变化。

我们认真对待向后兼容性，并努力确保您能够获得平稳的升级体验。

我们非常期待您的反馈；请打开一个 问题 有问题、错误或建议。

确定已安装的版本

如果你已经升级到最新版本，但没有看到任何你期待的新功能，那么你的python环境可能仍在使用旧版本。

您可以通过以下方式确定运行时使用的版本：

import mcp_servers
print(mcp_servers.__version__)

需求

Python 3.8或更高版本。

贡献

看 贡献文档.