클라이언트 트랜스포트

클라이언트 트랜스포트 (Client transports)

모든 Client는 트랜스포트(transport) — 실제로 메시지를 나르는 것 — 를 통해 서버와 대화합니다.

별도로 구성할 일은 없습니다. Client는 위치 인자 하나를 받고 그 타입에서 트랜스포트를 알아냅니다.

각각의 서버 쪽(mcp.run()이 하는 일, 여러분이 배포하는 것)은 Running your server 입니다.

Streamable HTTP

URL 문자열을 넘기면 Streamable HTTP를 얻습니다. 배포 뒤에 두는 트랜스포트이고 가장 먼저 손이 가는 것이죠.

from mcp import Client


async def main() -> None:
    async with Client("http://localhost:8000/mcp") as client:
        result = await client.list_tools()
        print([tool.name for tool in result.tools])

그게 운영 클라이언트의 전부입니다. Client는 URL을 streamable_http_client(...)로 감싸고, 그 위에 MCP가 필요로 하는 방식으로 구성된 httpx2.AsyncClient를 얹습니다. connect/write/pool에 30초 타임아웃, 읽기에 300초 타임아웃 — 서버가 응답 스트림을 열어 둘 수 있으니까요.

!!! check 생성한 Client는 연결되어 있지 않습니다. 생성은 트랜스포트를 고르기만 하고, async with가 그걸 여는 거죠. 들어가기 전에 연결에 손을 대면 SDK가 알려줍니다.

```text
RuntimeError: Client must be used within an async context manager
```

`Client("http://...")`를 썼을 때 아무것도 resolve·fetch·spawn되지 않았습니다. 그 줄은 공짜예요.

자신의 httpx2.AsyncClient 가져오기

Authorization 헤더, 쿠키, 프록시, mTLS, 다른 타임아웃이 필요해지는 순간, httpx2.AsyncClient를 직접 만들고 streamable_http_client에 넘기세요.

import httpx2

from mcp import Client
from mcp.client.streamable_http import streamable_http_client


async def main() -> None:
    async with httpx2.AsyncClient(
        headers={"Authorization": "Bearer ..."},
        timeout=httpx2.Timeout(30.0, read=300.0),
    ) as http_client:
        transport = streamable_http_client("http://localhost:8000/mcp", http_client=http_client)
        async with Client(transport) as client:
            result = await client.list_tools()
            print([tool.name for tool in result.tools])

두 가지를 주목하세요.

  • httpx2.AsyncClient는 여러분 것이므로 여러분이 들어가고 나갑니다. SDK는 자신이 만들지 않은 클라이언트를 절대 닫지 않아요.
  • streamable_http_client(url, http_client=...)는 트랜스포트를 반환하고, Client(transport)는 그걸 다른 것처럼 받아들입니다.

TLS 메모 하나. httpx2는 번들된 CA 목록이 아니라 운영체제 신뢰 저장소(truststore 경유)로 인증서를 검증합니다. 쓸 수 있는 시스템 CA 저장소가 없는 환경(일부 미니멀 컨테이너)에서는 표준 SSL_CERT_FILE/SSL_CERT_DIR 환경 변수를 설정하거나, httpx2.AsyncClient에 명시적 verify=ssl_context를 넘기세요(배경은 httpx and httpx-sse replaced by httpx2).

!!! warning streamable_http_client는 예전에 headers=와 timeout=을 직접 받았습니다. 이제는 그렇지 않아요. 파라미터는 url, http_client, terminate_on_close뿐입니다. 습관적으로 headers=에 손이 가면:

```text
TypeError: streamable_http_client() got an unexpected keyword argument 'headers'
```

HTTP 모양의 모든 것은 이제 넘겨주는 그 `httpx2.AsyncClient` 하나에 있습니다.

!!! info httpx2는 익숙한 httpx API를 유지합니다. httpx를 안다면 여기서 auth·프록시·이벤트 훅·재시도·연결 한도를 이미 아는 것이죠. SDK는 그 위에 아무것도 더하거나 빼지 않아요. 리다이렉트 처리만 빼고. OAuth도 여기서 꽂힙니다. httpx2.AsyncClient(auth=OAuthClientProvider(...)). 그 전체 흐름은 OAuth clients 입니다.

리다이렉트

트랜스포트는 여러분이 준 URL, 그리고 그 오리진에만 연결합니다.

  • 같은 scheme·host·port에 머무는 307/308 리다이렉트는 따르고, 같은 호스트의 http:// → https://도 따릅니다. 그게 보통의 /mcp → /mcp/ 후행 슬래시 리다이렉트를 덮습니다.

  • 그 밖의 어디로든 리다이렉트는 따르지 않습니다. 호출이 이렇게 실패합니다.

    MCPError: Redirect to https://other.example.com/mcp not followed; use that URL as the endpoint if it is the intended server

    그 URL이 의도한 서버라면 설정에 넣으세요. 아니라면 서버나 그 앞의 프록시가 잘못 구성된 것입니다.

이것은 넘겨주는 어떤 httpx2.AsyncClient에도 적용됩니다. MCP 요청에 대해서는 어느 방향으로든 follow_redirects 설정을 보지 않아요. SDK의 OAuth 프로바이더도 자기 요청에 같은 규칙을 적용합니다.

!!! tip Redirect to http://… not followed: it would downgrade this HTTPS endpoint to plain HTTP는 서버가 모르는 TLS 종단 프록시 뒤에 있고 http:// 리다이렉트를 내보내고 있다는 뜻입니다. 그건 서버 쪽에서 고쳐집니다(Deploy & scale), 아니면 메시지가 제안하는 정확한 https://…/ URL을 쓰면 됩니다.

stdio

stdio 서버는 서브프로세스입니다. 클라이언트가 그걸 띄우고 stdin에 JSON-RPC를 쓰고 stdout에서 JSON-RPC를 읽습니다. 데스크톱 호스트가 여러분 머신의 서버를 실행하는 방식이죠. 호스트 는 이 코드에 UI를 더한 것이고, Connect to a real host 는 호스트 쪽에서 본 같은 관계를 설정 파일로 보여줍니다.

StdioServerParameters로 프로세스를 기술하고 Client에 넘기세요.

from mcp import Client, StdioServerParameters

server = StdioServerParameters(
    command="uv",
    args=["run", "server.py"],
    env={"BOOKSHOP_API_KEY": "secret"},
)


async def main() -> None:
    async with Client(server) as client:
        result = await client.list_tools()
        print([tool.name for tool in result.tools])

블록에 들어가면 프로세스가 탄생합니다. 나가면 서브프로세스가 종료됩니다. stdin 닫고, 기다리고, 남아 있으면 kill. 직접 정리할 필요가 없어요.

자식의 stderr는 여러분의 stderr로 갑니다. 다른 데로 보내려면 stdio_client(mcp에서)로 트랜스포트를 직접 만들고 그걸 넘기세요. Client(stdio_client(server, errlog=log_file)).

!!! warning 자식은 여러분의 환경을 상속하지 않습니다. 최소 허용 목록(POSIX에서 HOME, LOGNAME, PATH, SHELL, TERM, USER)을 받아요. 여러분이 쓰지 않았을 수도 있는 프로세스에 민감한 것이 새지 않도록요.

API 키가 필요한 서버는 그걸 거기서 못 찾습니다. `env=`로 명시적으로 넘기세요. 그 변수들은 허용 목록 위에 병합됩니다. 위에서 `BOOKSHOP_API_KEY`가 하는 일이 그겁니다.

인메모리(In memory)

테스트에서는 배포할 것도, 띄울 것도 없습니다. 서버 객체 자체를 넘기세요.

from mcp import Client
from mcp.server import MCPServer

mcp = MCPServer("Bookshop")


@mcp.tool()
def search_books(query: str) -> str:
    """Search the catalog by title or author."""
    return f"Found 3 books matching {query!r}."


async def main() -> None:
    async with Client(mcp) as client:
        result = await client.call_tool("search_books", {"query": "dune"})
        print(result.structured_content)

서브프로세스도, 포트도, 와이어 위의 바이트도 없습니다. 클라이언트와 서버는 같은 프로세스의 객체 둘인데, 호출은 여전히 진짜 프로토콜 레이어를 통과합니다. search_books는 HTTP로 그랬을 것처럼 정확히 나열·검증·호출되죠. Testing 이 그 주위로 전체 패턴을 구축합니다.

같은 형태는 임베딩 API로도 두 배가 됩니다. 서버를 직접 만드는 애플리케이션은 네트워크 홉 없이 그 도구들을 호출할 수 있어요.

SSE

sse_client(url)(mcp.client.sse에서)는 Streamable HTTP가 대체한 HTTP 트랜스포트입니다. 같은 방식으로 감싸서(Client(sse_client("http://localhost:8000/sse"))) 여전히 그것을 말하는 서버와 대화하고, 그 위에 새로 만들지는 마세요.

Transport 프로토콜

Client에게 위의 전부는 같은 것입니다.

트랜스포트는 (read, write) 메시지 스트림 쌍을 yield하는 async 컨텍스트 매니저입니다. 공식적으로는 mcp.client의 Transport 프로토콜이죠. Client는 인자를 타입으로 resolve합니다. str은 streamable_http_client(url)이 되고, StdioServerParameters는 stdio_client(params)가 되며, 서버 객체는 인메모리로 연결되고, 그 밖의 것은 트랜스포트로 직접 들어갑니다. 그 마지막 규칙 덕분에 stdio_client(...), streamable_http_client(...), sse_client(...) 모두 같은 슬롯에 들어가고, 여러분이 직접 쓸 수도 있는 것입니다.

정리(Recap)

  • Client("http://.../mcp")(URL)는 운영 트랜스포트인 Streamable HTTP로 연결합니다.
  • 헤더·auth·프록시·타임아웃은 streamable_http_client(url, http_client=...)에 넘기는 httpx2.AsyncClient에 속합니다. headers= 키워드는 없어요.
  • 리다이렉트는 URL 자신의 오리진 안에서만(후행 슬래시 307/308) 그리고 같은 호스트의 http→https만 따릅니다. 그 밖에는 Redirect to … not followed로 실패하고, 최종 URL을 구성하세요.
  • stdio는 Client(StdioServerParameters(...))입니다. 자식 stderr를 돌리려고 할 때만 stdio_client(...)로 직접 감쌉니다.
  • 서브프로세스는 여러분의 환경이 아니라 허용 목록 환경을 받습니다. env=가 더합니다.
  • Client(mcp)(서버 객체)는 인메모리로 연결합니다. 테스트에서, 또는 서버를 만든 애플리케이션에 서버를 임베딩할 때 쓰세요.
  • 트랜스포트는 async with x as (read, write)할 수 있는 무엇이든입니다. Client는 서버 객체·URL·StdioServerParameters가 아닌 것을 그 프로토콜에 그대로 넘깁니다.
  • Client를 생성하는 것은 트랜스포트를 고르는 것. async with가 여는 것입니다.

트랜스포트가 열리면 양측이 프로토콜 버전에 합의해야 합니다. 평소엔 생각할 일이 없고, 필요할 때는 Protocol versions 페이지가 있습니다.

출처: Python SDK — Client transports

더 알아보기 (Learn more)