विषय पर बढ़ें

Client

मशीनी अनुवाद

यह page अंग्रेज़ी documentation से अपने-आप अनुवादित किया गया है, और अंग्रेज़ी page ही प्रामाणिक version है। अगर कुछ गलत लगे, तो अनुवाद page बताता है कि इसकी सूचना कैसे दें।

Client वह ज़रिया है जिससे Python program किसी MCP server से बात करता है।

यह एक object है जिसका एक ही lifecycle है: इसे बनाएँ, async with में enter करें, methods call करें। protocol का हर verb (tools की सूची लेना, किसी tool को call करना, resource पढ़ना, prompt render करना) इस पर एक async method है जो typed result लौटाता है।

आपका पहला client

client को बात करने के लिए server चाहिए। इस page का हर उदाहरण इसी Bookshop से connect करता है। इसे server.py के नाम से save करें और HTTP पर चलता छोड़ दें:

server.py
from pydantic import BaseModel

from mcp.server import MCPServer
from mcp.server.mcpserver.exceptions import ToolError
from mcp.types import Completion, CompletionArgument, CompletionContext, PromptReference, ResourceTemplateReference

mcp = MCPServer("Bookshop", instructions="Search the catalog before recommending a book.")

GENRES = ["fiction", "non-fiction", "poetry"]


class Book(BaseModel):
    title: str
    author: str
    year: int


@mcp.tool(title="Search the catalog")
def search_books(query: str, limit: int = 10) -> str:
    """Search the catalog by title or author."""
    return f"Found 3 books matching {query!r} (showing up to {limit})."


@mcp.tool()
def lookup_book(title: str) -> Book:
    """Look up a book by its exact title."""
    if title != "Dune":
        raise ToolError(f"No book titled {title!r} in the catalog.")
    return Book(title="Dune", author="Frank Herbert", year=1965)


@mcp.resource("catalog://genres")
def genres() -> list[str]:
    """The genres the catalog is organised by."""
    return GENRES


@mcp.resource("catalog://genres/{genre}")
def books_in_genre(genre: str) -> str:
    """Every title we stock in one genre."""
    return f"3 books filed under {genre}."


@mcp.prompt(title="Recommend a book")
def recommend(genre: str) -> str:
    """Ask for a recommendation in a genre."""
    return f"Recommend one {genre} book from the catalog and say why."


@mcp.completion()
async def complete_genre(
    ref: PromptReference | ResourceTemplateReference,
    argument: CompletionArgument,
    context: CompletionContext | None,
) -> Completion | None:
    return Completion(values=[genre for genre in GENRES if genre.startswith(argument.value)])
uv run mcp run server.py --transport streamable-http

इससे server http://localhost:8000/mcp पर serve होता है। client अपना अलग program है। इसे client.py के नाम से save करें और दूसरे terminal में python client.py चलाएँ:

client.py
import anyio

from mcp import Client


async def main() -> None:
    async with Client("http://localhost:8000/mcp") as client:
        print(client.server_info)
        print(client.server_capabilities)
        print(client.protocol_version)
        print(client.instructions)


if __name__ == "__main__":
    anyio.run(main)
  • Client("http://localhost:8000/mcp") को URL दिया गया है, इसलिए यह अभी शुरू किए गए server से Streamable HTTP पर connect करता है।
  • async with ही lifecycle है। इसमें enter करते ही connect और negotiate होता है; बाहर निकलते ही disconnect। कोई connect() / close() जोड़ी नहीं है, और block खत्म होने के बाद Client दोबारा इस्तेमाल नहीं हो सकता।
  • block के अंदर connection की जानकारी पहले से सादी properties के रूप में मौजूद है।

Client को क्या दे सकते हैं

Client एक positional argument लेता है और उसके type से transport तय करता है:

  • URL string (Client("http://localhost:8000/mcp")): Streamable HTTP, वह transport जिसके पीछे आप deploy करते हैं।
  • StdioServerParameters: वह command जो local subprocess के रूप में launch होता है, और जिससे उसके stdin और stdout के ज़रिए बात होती है।
  • transport: कोई भी चीज़ जिसे आप async with ... as (read, write) कर सकें, जैसे आपके अपने HTTP client के ऊपर streamable_http_client(url, http_client=...)
  • MCPServer (या low-level Server) instance: in-process connect होता है, न subprocess, न port। यह tests के लिए है, और Testing इसी पर आगे बढ़ता है।

इस page की बाकी हर चीज़ चारों में एक जैसी है। Headers, subprocesses, timeouts और Transport protocol का अपना अलग page है: Client transports

connected client पर क्या है

चार read-only properties, जो block में enter करते ही भर जाती हैं:

  • client.server_info: server की पहचान, या None अगर 2026 पीढ़ी का server इसे report नहीं करता (python-sdk servers default रूप से करते हैं)। यहाँ server_info.name "Bookshop" है, और server_info.version वही है जो server report करता है।
  • client.server_capabilities: server क्या कर सकता है (tools, resources, prompts, completions, ...)। जो capability server के पास नहीं है वह None होती है।
  • client.protocol_version: वह protocol version जिस पर दोनों पक्ष सहमत हुए। यहाँ यह "2026-07-28" है।
  • client.instructions: server की instructions= string, या None अगर उसने कोई set नहीं की।

आपने कोई protocol version नहीं चुना। default रूप से Client server को probe करता है और पुराने servers पर पुराने classic handshake पर लौट आता है, इसलिए एक ही client किसी भी पीढ़ी के server के साथ काम करता है। जब इसे नियंत्रित करने की ज़रूरत हो, पूरी जानकारी Protocol versions में है।

Tip

client.session अंदर का ClientSession है, low-level escape hatch। इस page की किसी भी चीज़ के लिए आपको इसकी ज़रूरत नहीं पड़ेगी।

tools की सूची लेना

client.py
import anyio

from mcp import Client


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


if __name__ == "__main__":
    anyio.run(main)

list_tools() एक ListToolsResult लौटाता है; tools .tools में हैं। हर एक वह पूरी definition है जो host किसी model को देगा। यह पहला है:

tool.name          # 'search_books'
tool.title         # 'Search the catalog'
tool.description   # 'Search the catalog by title or author.'

और tool.input_schema वह JSON Schema है जो server ने function के type hints से निकाला:

{
  "type": "object",
  "properties": {
    "query": {"title": "Query", "type": "string"},
    "limit": {"default": 10, "title": "Limit", "type": "integer"}
  },
  "required": ["query"],
  "title": "search_booksArguments"
}

UI को argument form दिखाने के लिए, और model को valid arguments बनाने के लिए, जो कुछ चाहिए वह सब इसी schema में है।

दूसरा tool, lookup_book, बिना title= के register हुआ था, इसलिए उसका tool.title None है।

Tip

title optional है, इसलिए किसी इंसान को tools दिखाने वाले UI को चुनना पड़ता है: title हो तो वही, नहीं तो namefrom mcp.shared.metadata_utils import get_display_name ठीक यही करता है, tools, resources, resource templates और prompts के लिए।

tool call करना

call_tool(name, arguments) tool चलाता है और आपको CallToolResult वापस देता है।

client.py
import anyio

from mcp import Client
from mcp.types import TextContent


async def main() -> None:
    async with Client("http://localhost:8000/mcp") as client:
        result = await client.call_tool("lookup_book", {"title": "Dune"})

        for block in result.content:
            if isinstance(block, TextContent):
                print(block.text)

        print(result.structured_content)
        print(result.is_error)


if __name__ == "__main__":
    anyio.run(main)

server का lookup_book एक Pydantic Book लौटाता है। client को यह दिखता है:

result.content             # [TextContent(type='text', text='{\n  "title": "Dune",\n  "author": "Frank Herbert",\n  "year": 1965\n}')]
result.structured_content  # {'title': 'Dune', 'author': 'Frank Herbert', 'year': 1965}
result.is_error            # False

एक return value, पढ़ने की तीन चीज़ें। हर एक को पढ़ने वाला अलग है।

content: जो model पढ़ता है

content content blocks की एक list है, और content block एक union है: TextContent, ImageContent, AudioContent, ResourceLink, या EmbeddedResource। एक tool कई blocks लौटा सकता है, अलग-अलग तरह के।

इसीलिए main block.text को छूने से पहले isinstance(block, TextContent) से narrow करता है। ध्यान दें कि isinstance के बाहर कहीं .text नहीं है: type checker इसकी अनुमति नहीं देगा, क्योंकि ImageContent में .data है, .text नहीं। tool आपको क्या भेज सकता है, इस बारे में union ईमानदार है; आपका code भी होना चाहिए।

structured_content: जो आपका application पढ़ता है

structured_content tool की return value JSON के रूप में है, जो tool के declared output_schema से मेल खाती है। न string parsing, न अंदाज़ा।

जब दोनों मौजूद हों तो वे जानबूझकर एक ही बात दो बार कहते हैं: content model के लिए है, structured_content code के लिए। structured वाला हिस्सा कहाँ से आता है, और उसे कैसे नियंत्रित करें, यह Structured output page पर है।

is_error: tool fail हुआ या नहीं

जो tool raise करता है वह आपके client में raise नहीं होता। वह is_error=True के साथ एक साधारण result के रूप में लौटता है।

Check

lookup_book से "Solaris" माँगें (ऐसा title जो catalog में नहीं है) और function ToolError raise करता है। call फिर भी सामान्य रूप से लौटता है:

result.is_error            # True
result.content             # [TextContent(type='text', text="Error executing tool lookup_book: No book titled 'Solaris' in the catalog.")]
result.structured_content  # None

ToolError का message content में पहुँचा, जहाँ model उसे पढ़कर दोबारा कोशिश कर सकता है। यह जानबूझकर है: tool error बातचीत का हिस्सा है, crash नहीं। (अगर tool किसी और exception से crash हुआ होता, तो content में सिर्फ़ Error executing tool lookup_book लिखा होता।) structured_content पर भरोसा करने से पहले हमेशा is_error देखें।

Warning

is_error=True सिर्फ़ आपके अपने raise तक सीमित नहीं है। ऐसा tool माँगें जो server के पास है ही नहीं (call_tool("does_not_exist", {})) और कुछ raise नहीं होता। आपको वही shape वापस मिलता है, is_error=True और content में Unknown tool: does_not_existClient का कोई method MCPError तभी raise करता है जब server result की जगह JSON-RPC error से जवाब दे, और server कब क्या भेजता है यह errors संभालना में बताया गया है।

Resources

resource verbs जोड़ियों में आते हैं: सूची लेने के दो तरीके, पढ़ने का एक।

client.py
import anyio

from mcp import Client
from mcp.types import TextResourceContents


async def main() -> None:
    async with Client("http://localhost:8000/mcp") as client:
        listed = await client.list_resources()
        print([resource.uri for resource in listed.resources])

        templates = await client.list_resource_templates()
        print([template.uri_template for template in templates.resource_templates])

        result = await client.read_resource("catalog://genres/poetry")
        for contents in result.contents:
            if isinstance(contents, TextResourceContents):
                print(contents.text)


if __name__ == "__main__":
    anyio.run(main)
  • list_resources() concrete resources लौटाता है, जिनका URI तय है। यहाँ: ['catalog://genres']
  • list_resource_templates() parameterised वाले लौटाता है। यहाँ: ['catalog://genres/{genre}']। ये दो अलग सूचियाँ हैं क्योंकि template तब तक पढ़ा नहीं जा सकता जब तक आप उसे भर न दें।
  • read_resource(uri) एक सादा str URI लेता है और दोनों पर काम करता है: "catalog://genres/poetry" दें और server उसे template से match कर लेता है।

read_resource contents लौटाता है, TextResourceContents या BlobResourceContents की सूची। वही तरीका जो tool content का है: isinstance से narrow करें, फिर .text (या .blob) पढ़ें।

client को यह भी बताया जा सकता है कि कोई resource कब बदला। 2025 पीढ़ी के connections पर यह subscribe_resource(uri) / unsubscribe_resource(uri) है - methods की ऐसी जोड़ी जिसे MCPServer implement नहीं करता, इसलिए 2026-07-28 wire पर (जहाँ ये verbs अब मौजूद नहीं हैं) request का जवाब -32601, Method not found आता है। 2026 में इसकी जगह subscriptions/listen stream है, जिसे MCPServer serve करता है - वहाँ server_capabilities.resources.subscribe True है - और उसे client.listen(...) से consume करना इस section का Subscriptions page है।

Prompts

client.py
import anyio

from mcp import Client


async def main() -> None:
    async with Client("http://localhost:8000/mcp") as client:
        listed = await client.list_prompts()
        print(listed.prompts)

        result = await client.get_prompt("recommend", {"genre": "poetry"})
        for message in result.messages:
            print(message.role, message.content)


if __name__ == "__main__":
    anyio.run(main)

list_prompts() बताता है कि server क्या देता है और हर prompt को क्या चाहिए:

prompt.name        # 'recommend'
prompt.title       # 'Recommend a book'
prompt.arguments   # [PromptArgument(name='genre', required=True)]

get_prompt(name, arguments) उसे render करता है। arguments dict str -> str है: prompt arguments हमेशा strings होते हैं। result messages है, PromptMessage की सूची, जिनमें हर एक का role और एक content block है:

message.role     # 'user'
message.content  # TextContent(type='text', text='Recommend one poetry book from the catalog and say why.')

host ये messages सीधे model को दे देता है। पूरा feature बस इतना ही है।

Completions

जिस server में completion handler हो वह user के type करते-करते prompt और resource-template arguments autocomplete कर सकता है।

client.py
import anyio

from mcp import Client
from mcp.types import PromptReference


async def main() -> None:
    async with Client("http://localhost:8000/mcp") as client:
        result = await client.complete(
            ref=PromptReference(type="ref/prompt", name="recommend"),
            argument={"name": "genre", "value": "p"},
        )
        print(result.completion.values)


if __name__ == "__main__":
    anyio.run(main)
  • ref बताता है कि आप कौन-सा prompt या template भर रहे हैं: PromptReference या ResourceTemplateReference
  • argument {"name": ..., "value": ...} है: argument और user ने अब तक जो type किया है।

जवाब result.completion.values में है। "p" type करें और server ['poetry'] लौटाता है। server वाला पक्ष, और handler पहले से भरे बाकी arguments का इस्तेमाल अपने सुझाव कम करने के लिए कैसे करता है, यह Completions page पर है।

Pagination

हर list_* method एक cursor= keyword लेता है और हर result में next_cursor होता है। जब next_cursor None हो, आपके पास सब कुछ है।

client.py
import anyio

from mcp import Client
from mcp.types import Tool


async def list_all_tools(client: Client) -> list[Tool]:
    tools: list[Tool] = []
    cursor: str | None = None
    while True:
        page = await client.list_tools(cursor=cursor)
        tools.extend(page.tools)
        if page.next_cursor is None:
            return tools
        cursor = page.next_cursor


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


if __name__ == "__main__":
    anyio.run(main)

list_all_tools हर server के साथ सही है। MCPServer सब कुछ एक ही page में लौटाता है, इसलिए next_cursor None होता है और loop एक बार चलता है, यही वजह है कि ज़्यादातर code इसे कभी लिखता ही नहीं। जो servers सच में page करते हैं, और cursors जिन नियमों का पालन करते हैं, वे Pagination में हैं।

tests में

इस page की हर client.py HTTP के ज़रिए server.py तक पहुँची। test में आप network छोड़ देते हैं और Client को server object ही दे देते हैं: from server import mcp, फिर Client(mcp)। न process, न port, और ऊपर का हर method वैसे ही काम करता है।

इसी के लिए एक constructor flag बना है: Client(mcp, raise_exceptions=True)। इसका असर सिर्फ़ in-process connections पर होता है, और Testing वह page है जो इसे समझाता है और इसके चारों ओर पूरा pattern बनाता है।

सारांश

  • Client(x) URL string से Streamable HTTP पर connect होता है, StdioServerParameters के लिए subprocess launch करता है, transport में सीधे enter करता है, और tests में server object ही ले लेता है।
  • async with ही पूरा lifecycle है। इसके अंदर server_capabilities और protocol_version पहले से भरे होते हैं; server दे तो server_info और instructions भी।
  • list_tools() आपको हर tool का name, title, description और input_schema देता है।
  • call_tool() model के लिए content, आपके code के लिए structured_content, और is_error लौटाता है। raise करने वाला tool एक result है, exception नहीं।
  • content block types का union है; पढ़ने से पहले isinstance से narrow करें।
  • list_resources / list_resource_templates / read_resource, list_prompts / get_prompt, और complete बाकी verbs पूरे करते हैं।
  • हर list_* cursor= लेता है; next_cursor के None होने तक loop करें।

server client से जो चीज़ें माँग सकता है, और आप उनका जवाब कैसे देते हैं, वह Client callbacks है।