python版sdkconnect_to_server无法传递header的Authorization

[loli0123456789]

python版sdkconnect_to_server无法传递header的Authorization
但是mcp官方的sse_client是可以传递url、header的。

这里建议可以完善下。

[loli0123456789]

@zhangairku
是百度appbuilder的connect_to_server不能传递Authorization

百度MCPClient代码

# from appbuilder.mcp_server.client import MCPClient

async def connect_to_server(
        self, server_script_path: str = None, service_url: str = None, env: dict = None
    ):
        """Connect to an MCP server

        Args:
            server_script_path: Path to the server script (.py or .js)
            service_url: URL of the service
        """
        if server_script_path is not None:
            await self._connect_to_stdio_server(server_script_path, env=env)
        elif service_url is not None:
            await self._connect_to_sse_server(service_url)
        else:
            raise ValueError(
                "Either server_script_path or service_url must be provided."
            )

async def _connect_to_sse_server(self, service_url: str):
        self._streams_context = sse_client(url=service_url)
        streams = await self._streams_context.__aenter__()
        self._session_context = ClientSession(*streams)
        self.session: ClientSession = await self._session_context.__aenter__()
        await self.session.initialize()
        response = await self.session.list_tools()
        tools = response.tools
        self.tools = tools
        logger.info(
            "\nConnected to server with tools:" +
            str([tool.name for tool in tools])
        )
        self._store_session_and_tools(service_url)

self._streams_context = sse_client(url=service_url)
可以看到这里只是传递了service_url

mcp官方sse_client

人家mcp官方的是可以传递Authorization的，通过headers参数即可实现

async def sse_client(
    url: str,
    headers: dict[str, Any] | None = None,
    timeout: float = 5,
    sse_read_timeout: float = 60 * 5,
    httpx_client_factory: McpHttpClientFactory = create_mcp_http_client,
    auth: httpx.Auth | None = None,
):
    """
    Client transport for SSE.

    `sse_read_timeout` determines how long (in seconds) the client will wait for a new
    event before disconnecting. All other HTTP operations are controlled by `timeout`.

    Args:
        url: The SSE endpoint URL.
        headers: Optional headers to include in requests.
        timeout: HTTP timeout for regular operations.
        sse_read_timeout: Timeout for SSE read operations.
        auth: Optional HTTPX authentication handler.
    """
    read_stream: MemoryObjectReceiveStream[SessionMessage | Exception]
    read_stream_writer: MemoryObjectSendStream[SessionMessage | Exception]

    write_stream: MemoryObjectSendStream[SessionMessage]
    write_stream_reader: MemoryObjectReceiveStream[SessionMessage]

    read_stream_writer, read_stream = anyio.create_memory_object_stream(0)
    write_stream, write_stream_reader = anyio.create_memory_object_stream(0)

    async with anyio.create_task_group() as tg:
        try:
            logger.debug(f"Connecting to SSE endpoint: {remove_request_params(url)}")
            async with httpx_client_factory(headers=headers, auth=auth) as client:
                async with aconnect_sse(
                    client,
                    "GET",
                    url,
                    timeout=httpx.Timeout(timeout, read=sse_read_timeout),
                ) as event_source:
                    event_source.response.raise_for_status()
                    logger.debug("SSE connection established")

                    async def sse_reader(
                        task_status: TaskStatus[str] = anyio.TASK_STATUS_IGNORED,
                    ):
                        try:
                            async for sse in event_source.aiter_sse():
                                logger.debug(f"Received SSE event: {sse.event}")
                                match sse.event:
                                    case "endpoint":
                                        endpoint_url = urljoin(url, sse.data)
                                        logger.debug(f"Received endpoint URL: {endpoint_url}")

                                        url_parsed = urlparse(url)
                                        endpoint_parsed = urlparse(endpoint_url)
                                        if (
                                            url_parsed.netloc != endpoint_parsed.netloc
                                            or url_parsed.scheme != endpoint_parsed.scheme
                                        ):
                                            error_msg = (
                                                "Endpoint origin does not match " f"connection origin: {endpoint_url}"
                                            )
                                            logger.error(error_msg)
                                            raise ValueError(error_msg)

                                        task_status.started(endpoint_url)

                                    case "message":
                                        try:
                                            message = types.JSONRPCMessage.model_validate_json(  # noqa: E501
                                                sse.data
                                            )
                                            logger.debug(f"Received server message: {message}")
                                        except Exception as exc:
                                            logger.error(f"Error parsing server message: {exc}")
                                            await read_stream_writer.send(exc)
                                            continue

                                        session_message = SessionMessage(message)
                                        await read_stream_writer.send(session_message)
                                    case _:
                                        logger.warning(f"Unknown SSE event: {sse.event}")
                        except Exception as exc:
                            logger.error(f"Error in sse_reader: {exc}")
                            await read_stream_writer.send(exc)
                        finally:
                            await read_stream_writer.aclose()

                    async def post_writer(endpoint_url: str):
                        try:
                            async with write_stream_reader:
                                async for session_message in write_stream_reader:
                                    logger.debug(f"Sending client message: {session_message}")
                                    response = await client.post(
                                        endpoint_url,
                                        json=session_message.message.model_dump(
                                            by_alias=True,
                                            mode="json",
                                            exclude_none=True,
                                        ),
                                    )
                                    response.raise_for_status()
                                    logger.debug("Client message sent successfully: " f"{response.status_code}")
                        except Exception as exc:
                            logger.error(f"Error in post_writer: {exc}")
                        finally:
                            await write_stream.aclose()

                    endpoint_url = await tg.start(sse_reader)
                    logger.debug(f"Starting post writer with endpoint URL: {endpoint_url}")
                    tg.start_soon(post_writer, endpoint_url)

                    try:
                        yield read_stream, write_stream
                    finally:
                        tg.cancel_scope.cancel()
        finally:
            await read_stream_writer.aclose()
            await write_stream.aclose()

---

能把你使用connect_to_server传递Authorization失败的源代码展示一下吗？

[loli0123456789]

这个仓库不活跃，提个问题，官方也没人管。。。

[zhangairku]

@loli0123456789 sdk的mcp调用测试案例是

app-builder/python/tests/test_mcp_app_server_stdio.py

Lines 31 to 32 in 8fb01a4

	mcp_client = MCPClient()
	env = {"APPBUILDER_TOKEN": os.getenv("APPBUILDER_TOKEN")}

，这里的没法符合你的要求吗？还是说你需要的是jwt token那种直接放在http请求头的Authorization呢？

[zhangairku]

@loli0123456789 我们这边打算重构仓库代码的，sdk现在的维护人员不是最开始的人了，所以不是很熟练，非常抱歉。如果有问题，可以到微信群中建议，可以更及时被回复