Obsah
1. Krátké zopakování: MCP server s definovaným nástrojem
2. MCP klient volající nástroj spouštěný z MCP serveru
3. Úplný zdrojový kód klienta, který získá hodnoty faktoriálů a vytiskne je
4. MCP server se zdrojem se seznamem knih autorů
5. MCP klient, který přečte zvolený zdroj a získá z něj strukturovaná data
6. Odlišný oddělovač selektorů
7. Chování MCP serveru po příchodu většího množství požadavků
8. Přístup ke „zpomalenému“ MCP serveru z MCP klienta
9. Souběžný přístup ke „zpomalenému“ MCP serveru
10. Asynchronní souběžný přístup ke „zpomalenému“ MCP serveru
11. Podpora rastrových obrázků v balíčku mcp
12. MCP server posílající rastrový obrázek
13. Dekódování obrázku na straně MCP klienta
14. Repositář s demonstračními příklady
15. Příloha: články o knihovně PIL/Pillow
1. Krátké zopakování: MCP server s definovaným nástrojem
V úvodním článku o MCP (Model Context Protocol) jsme se seznámili se základními vlastnosti tohoto protokolu. Ukázali jsme si taktéž knihovnu mcp určenou pro programovací jazyk Python, která umožňuje tvorbu MCP serverů i MCP klientů. A nezapomněli jsme ani na MCP Inspector, což je vývojářský nástroj zajišťující připojení k MCP serverům (nebo i k jejich spuštění) a interaktivní komunikaci se serverem.
Pro krátké zopakování si znovu ukažme zdrojový kód MCP serveru, ve kterém je definován jeden zdroj (resource) bez vyžadovaných selektorů, dále zdroj se selektorem (konkrétně se jménem klienta) a konečně i nástroj (tool), který na základě zadaného vstupu vypočítá hodnotu faktoriálu.
Zdrojový kód serveru vypadá takto:
"""MCP server se zdrojem, dynamickým zdrojem a definicí nástroje (tool)."""
from mcp.server.fastmcp import FastMCP
# konstrukce serveru
mcp = FastMCP("Test")
@mcp.tool()
def factorial(n: int) -> int:
"""Výpočet faktoriálu ve smyčce."""
f = 1
for x in range(1, n + 1):
f *= x
return f
@mcp.resource("pozdrav://")
def pozdrav1() -> str:
"""Odpověď s pozdravem."""
return "Hello, dear client"
@mcp.resource("pozdrav://{name}")
def pozdrav2(name: str) -> str:
"""Odpověď s osobním pozdravem."""
return f"Hello, {name}"
# přímé spuštění serveru v režimu SSE (Server-Sent Events)
if __name__ == "__main__":
mcp.run(transport="sse")
2. MCP klient volající nástroj spouštěný z MCP serveru
Podívejme se nyní na to, jaká je základní struktura jednoduchého MCP klienta, který zvolený nástroj (tool) zavolá, předá mu všechny požadované argumenty, počká na přijaté výsledky a nakonec získá výslednou hodnotu poslanou klientem. Klient si sám zajistí spuštění MCP serveru a navázání komunikace s transportem STDIO:
# parametry pro spuštění MCP serveru
server_params = StdioServerParameters(
command="python", # spustí se tento příkaz
args=["mcp_server_5.py"], # a předají se mu následující parametry
env=None, # lze definovat i proměnné prostředí
)
Spuštění serveru (a na konci jeho ukončení) se děje v rámci správce kontextu (context manager):
async with stdio_client(server_params) as (read, write):
...
...
...
Nejprve získáme objekt představující sezení (session obsahující kontextové informace o navázaném připojení; sezení inicializujeme:
async with ClientSession(read, write) as session:
await session.initialize()
Mimochodem, k objektu typu ClientSession si samozřejmě můžete zobrazit nápovědu přímo z REPLu Pythonu:
>>> from mcp import ClientSession, StdioServerParameters >>> c=ClientSession(None, None) >>> help(c) class ClientSession(mcp.shared.session.BaseSession) | ClientSession(read_stream: anyio.streams.memory.MemoryObjectReceiveStream[mcp.types.JSONRPCMessage | Exception], write_stream: anyio.streams.memory.MemoryObjectSendStream[mcp.types.JSONRPCMessage], read_timeout_seconds: datetime.timedelta | None = None, sampling_callback: mcp.client.session.SamplingFnT | None = None, list_roots_callback: mcp.client.session.ListRootsFnT | None = None, logging_callback: mcp.client.session.LoggingFnT | None = None) -> None | | Method resolution order: | ClientSession | mcp.shared.session.BaseSession | typing.Generic | builtins.object |
Samotné volání nástroje, který je spouštěný přes MCP server, je v MCP klientovi provedeno asynchronně, ostatně jako všechna ostatní volání MCP serveru (zdroje atd.):
data = await session.call_tool("factorial")
Typicky je však nutné volanému nástroji předat argumenty, protože se vlastně nejedná o nic jiného, než o vzdálené volání kódu (RPC). Argumenty se specifikují v keyword parametru nazvaném arguments, jehož hodnotou je mapa obsahující jako klíče jména argumentů a hodnoty jsou (nepřekvapivě) předávané hodnoty. To je umožněno, protože hlavička metody call_tool vypadá následovně:
async call_tool(name: str, arguments: dict[str, typing.Any] | None = None)
-> mcp.types.CallToolResult method of mcp.client.session.ClientSession instance
Send a tools/call request.
V případě výpočtu faktoriálu tedy volání nástroje může vypadat následovně:
data = await session.call_tool("factorial", arguments={"n": n})
Můžeme se například pokusit o získání faktoriálů 0! až 10! (včetně) s následným výpisem získaných výsledků:
for n in range(11):
# zavolání nástroje
data = await session.call_tool("factorial", arguments={"n": n})
factorial = data.content[0].text
print(n, factorial)
3. Úplný zdrojový kód klienta, který získá hodnoty faktoriálů a vytiskne je
Úplný zdrojový kód MCP klienta popsaného v předchozí kapitole, který dokáže spustit MCP server, komunikovat s ním přes STDIO (standardní vstupy a výstupy) a získat hodnoty faktoriálů vypočtené nástrojem (tool) spouštěným z MCP serveru, vypadá ve své nejjednodušší podobě následovně:
"""MCP klient, který zavolá nástroj."""
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
# parametry pro spuštění MCP serveru
server_params = StdioServerParameters(
command="python", # spustí se tento příkaz
args=["mcp_server_5.py"], # a předají se mu následující parametry
env=None, # lze definovat i proměnné prostředí
)
async def run():
"""Realizace klienta."""
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
for n in range(11):
# zavolání nástroje
data = await session.call_tool("factorial", arguments={"n": n})
factorial = data.content[0].text
print(n, factorial)
# přímé spuštění klienta
if __name__ == "__main__":
import asyncio
asyncio.run(run())
Klienta běžným způsobem spustíme příkazem python jméno_skriptu.py a získáme potřebné výsledky. O inicializaci MCP serveru se již nemusíme starat, protože si klient server spustí automaticky. Výsledky by měly vypadat takto:
0 1 1 1 2 2 3 6 4 24 5 120 6 720 7 5040 8 40320 9 362880 10 3628800
V případě, že bychom ovšem vypisovali přímo odpovědi serveru, dostali bychom textovou podobu objektu, který je zpracován protokolem MCP a převeden do podoby Pythonovského objektu knihovnou mcp (to však většinou není zapotřebí):
meta=None content=[TextContent(type='text', text='1', annotations=None)] isError=False meta=None content=[TextContent(type='text', text='1', annotations=None)] isError=False meta=None content=[TextContent(type='text', text='2', annotations=None)] isError=False meta=None content=[TextContent(type='text', text='6', annotations=None)] isError=False meta=None content=[TextContent(type='text', text='24', annotations=None)] isError=False meta=None content=[TextContent(type='text', text='120', annotations=None)] isError=False meta=None content=[TextContent(type='text', text='720', annotations=None)] isError=False meta=None content=[TextContent(type='text', text='5040', annotations=None)] isError=False meta=None content=[TextContent(type='text', text='40320', annotations=None)] isError=False meta=None content=[TextContent(type='text', text='362880', annotations=None)] isError=False meta=None content=[TextContent(type='text', text='3628800', annotations=None)] isError=False
4. MCP server se zdrojem se seznamem knih autorů
V dnešním druhém demonstračním příkladu si připomeneme, jak by mohl vypadat jednoduchý MCP server, který pro zadaného autora (jméno, příjmení) vrací seznam knih. Jedná se o triviální implementaci bez databáze, takže se pro všechny autory vrací stejný seznam. To však pro základní testy nemusí být na škodu. Namísto nástroje (tool) v tomto případě použijeme zdroj. Ten je na straně serveru definován pod jménem author:// se selektorem name a surname, přičemž tyto selektory jsou odděleny pomlčkou (nemusíme tedy používat jen lomítka):
"""Jednoduchý MCP server s jediným definovaným dynamickým zdrojem."""
from mcp.server.fastmcp import FastMCP
# konstrukce serveru
mcp = FastMCP("Test")
@mcp.resource("author://{name}-{surname}")
def author(name: str, surname: str) -> list:
"""Knihy od vybraného autora."""
return [
{"name": name, "surname": surname, "title": "Foo", "year": 1900},
{"name": name, "surname": surname, "title": "Bar", "year": 2005},
{"name": name, "surname": surname, "title": "Baz", "year": 2025},
]
Tento server je možné spustit jak se STDIO transportem, tak i s SSE. Spustit ho můžeme nástrojem mcp nebo přímo z MCP klienta.
5. MCP klient, který přečte zvolený zdroj a získá z něj strukturovaná data
Nyní bude zajímavé zjistit, jak vlastně vypadá MCP klient, který dokáže přečíst a zpracovat jména knih napsaná vybraným autorem. Jediná zajímavá (resp. nová) část se týká zpracování dat, které nám knihovna mcp vrátí. Proto po spuštění klienta a přečtení odpovědi ze serveru zobrazíme jak „surová“ data, která byla přečtena a vrácena klientovi, tak i vlastní text (ano, čistý text!), s JSON odpovědí. O parsing JSONu se musíme postarat programově, což je zvláštní, ale současná verze knihovny mcp tuto vlastnost nepodporuje (je otevřena issue):
"""MCP klient, který přečte zvolený zdroj a získá z něj strukturovaná data."""
import json
from pprint import pprint
from mcp import ClientSession
from mcp.client.sse import sse_client
async def run():
"""Realizace klienta."""
async with sse_client(url="http://localhost:8000/sse") as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
# přečtení zdroje
data = await session.read_resource("author://john-doe")
print("Data returned:", data)
print("Type:", type(data))
text = data.contents[0].text
print("Text:", text)
deserialized = json.loads(text)
print("Deserialized:")
pprint(deserialized)
# přímé spuštění klienta
if __name__ == "__main__":
import asyncio
asyncio.run(run())
Po spuštění MCP klienta se nejdříve zobrazí celý objekt, který byl vrácen metodou session.read_resource. Tento objekt obsahuje sekvenci objektů typu TextResourceContents:
Data returned: meta=None contents=[TextResourceContents(uri=AnyUrl('author://john-doe'), mimeType='application/json', text='[{"name": "john", "surname": "doe", "title": "Foo", "year": 1900}, {"name": "john", "surname": "doe", "title": "Bar", "year": 2005}, {"name": "john", "surname": "doe", "title": "Baz", "year": 2025}]')]
Typ objektu:
Type: <class 'mcp.types.ReadResourceResult'>
Vlastní text přečtený z objektu je skutečně text:
Text: [{"name": "john", "surname": "doe", "title": "Foo", "year": 1900}, {"name": "john", "surname": "doe", "title": "Bar", "year": 2005}, {"name": "john", "surname": "doe", "title": "Baz", "year": 2025}]
Teprve po explicitním volání json.loads() (tedy po deserializaci) získáme z textu strukturovaná data se seznamem knih:
Deserialized:
[{'name': 'john', 'surname': 'doe', 'title': 'Foo', 'year': 1900},
{'name': 'john', 'surname': 'doe', 'title': 'Bar', 'year': 2005},
{'name': 'john', 'surname': 'doe', 'title': 'Baz', 'year': 2025}]
6. Odlišný oddělovač selektorů
Samozřejmě můžeme server překonfigurovat tak, aby byly jednotlivé selektory zdroje odděleny lomítky a nikoli pomlčkami. Postačuje k tomu velmi malá změna. Dekorátor zdroje:
@mcp.resource("author://{name}-{surname}")
pozměníme na:
@mcp.resource("author://{name}/{surname}", mime_type="application/json")
Upravený zdrojový kód serveru vypadá následovně:
"""Jednoduchý MCP server s jediným definovaným dynamickým zdrojem."""
from mcp.server.fastmcp import FastMCP
# konstrukce serveru
mcp = FastMCP("Test")
@mcp.resource("author://{name}/{surname}", mime_type="application/json")
def author(name: str, surname: str) -> list:
"""Knihy od vybraného autora."""
return [
{"name": name, "surname": surname, "title": "Foo", "year": 1900},
{"name": name, "surname": surname, "title": "Bar", "year": 2005},
{"name": name, "surname": surname, "title": "Baz", "year": 2025},
]
# přímé spuštění serveru v režimu SSE (Server-Sent Events)
if __name__ == "__main__":
mcp.run(transport="sse")
Samozřejmě je nutné změnit i klienta. Konkrétně se jedná o nahrazení řádku:
data = await session.read_resource("author://john-doe")
za řádek:
data = await session.read_resource("author://john/doe")
Opět si pro úplnost ukažme, jak klient vypadá po této nepatrné úpravě:
"""MCP klient, který přečte zvolený zdroj a získá z něj strukturovaná data."""
import json
from pprint import pprint
from mcp import ClientSession
from mcp.client.sse import sse_client
async def run():
"""Realizace klienta."""
async with sse_client(url="http://localhost:8000/sse") as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
# přečtení zdroje
data = await session.read_resource("author://john/doe")
print("Data returned:", data)
print("Type:", type(data))
text = data.contents[0].text
print("Text:", text)
deserialized = json.loads(text)
print("Deserialized:")
pprint(deserialized)
# přímé spuštění klienta
if __name__ == "__main__":
import asyncio
asyncio.run(run())
7. Chování MCP serveru po příchodu většího množství požadavků
V dalším zkoumání možností MCP se pokusíme zjistit, jakým způsobem MCP server reaguje na situaci, kdy dostane větší množství požadavků. Takový server tedy budeme muset spustit s transportem SSE a nikoli STDIO, protože komunikace přes STDIO je určena pro komunikaci pouze dvou procesů. Abychom odsimulovali nějakou složitější činnost MCP serveru, vložíme jak do volání nástroje (tool), tak i pro přístup ke zdrojům (resource) volání time.sleep s přibližně půlsekundovým intervalem. Tím budeme simulovat déletrvající operace. Nová podoba implementace MCP serveru bude vypadat následovně:
"""MCP server se zdrojem, dynamickým zdrojem a definicí nástroje (tool)."""
from time import sleep
from mcp.server.fastmcp import FastMCP
# konstrukce serveru
mcp = FastMCP("Test")
SLEEP_AMOUNT=0.5
@mcp.tool()
def factorial(n: int) -> int:
"""Výpočet faktoriálu ve smyčce."""
print("Factorial computation started")
f = 1
for x in range(1, n + 1):
f *= x
sleep(SLEEP_AMOUNT)
print("Factorial computation finished")
return f
@mcp.resource("pozdrav://")
def pozdrav1() -> str:
"""Odpověď s pozdravem."""
print("Resource preparation started")
sleep(SLEEP_AMOUNT)
print("Resource preparation finished")
return "Hello, dear client!"
@mcp.resource("pozdrav://{name}")
def pozdrav2(name: str) -> str:
"""Odpověď s osobním pozdravem."""
print("Resource preparation started")
sleep(SLEEP_AMOUNT)
print("Resource preparation finished")
return f"Hello, {name}!"
# přímé spuštění serveru v režimu SSE (Server-Sent Events)
if __name__ == "__main__":
mcp.run(transport="sse")
8. Přístup ke „zpomalenému“ MCP serveru z MCP klienta
Abychom si ověřili, jak rychle nebo pomalu přístup k MCP serveru probíhá, přidáme do MCP klienta výpočet celkového času nutného pro přečtení obou zdrojů i výsledku vypočteného volaným nástrojem. Je to snadné:
"""MCP klient, který přečte zvolený zdroj a získá z něj data."""
from time import time
from mcp import ClientSession, StdioServerParameters
from mcp.client.sse import sse_client
async def run():
"""Realizace klienta."""
async with sse_client(url="http://localhost:8000/sse") as (read, write):
async with ClientSession(read, write) as session:
t1 = time()
print("Client initialization")
print()
await session.initialize()
# přečtení zdroje bez selektoru
data = await session.read_resource("pozdrav://")
print("Data returned:", data)
print("Type:", type(data))
text = data.contents[0].text
print("Text:", text)
print()
# přečtení zdroje se selektorem
data = await session.read_resource("pozdrav://Pavel")
print("Data returned:", data)
print("Type:", type(data))
text = data.contents[0].text
print("Text:", text)
print()
# zavolání nástroje
data = await session.call_tool("factorial", arguments={"n": 10})
factorial = data.content[0].text
print("10!=", factorial)
print()
t2 = time()
difftime = t2 - t1
print(f"Client has finished in {difftime} seconds")
print()
# přímé spuštění klienta
if __name__ == "__main__":
import asyncio
asyncio.run(run())
Takto může vypadat běh klienta:
Client initialization
Data returned: meta=None contents=[TextResourceContents(uri=AnyUrl('pozdrav://'), mimeType='text/plain', text='Hello, dear client!')]
Type: <class 'mcp.types.ReadResourceResult'>
Text: Hello, dear client!
Data returned: meta=None contents=[TextResourceContents(uri=AnyUrl('pozdrav://Pavel'), mimeType='text/plain', text='Hello, Pavel!')]
Type: <class 'mcp.types.ReadResourceResult'>
Text: Hello, Pavel!
10!= 3628800
Client has finished in 1.5324900150299072 seconds
Povšimněte si, že jeden MCP klient skončil skutečně za cca 1,5 sekundy (tedy co jedno volání, to přibližně 0,5 sekundy).
9. Souběžný přístup ke „zpomalenému“ MCP serveru
Nyní se pokusíme o souběžné spuštění deseti MCP klientů. Jsou možné dva výsledky:
- Každý MCP klient bude se obsloužen souběžně a tedy každý skončí přibližně po 1,5 sekundě
- Obsluha bude prováděna sekvenčně a časy budou mnohem delší – poslední klient by měl skončit po 10×1,5=15 sekundách
Princip spuštění:
for i in `seq 10`
do
echo "Starting client #$i"
python mcp_client_8.py > "$i.txt" &
done
Výsledek pro nejrychlejšího MCP klienta:
Client initialization
Data returned: meta=None contents=[TextResourceContents(uri=AnyUrl('pozdrav://'), mimeType='text/plain', text='Hello, dear client!')]
Type: <class 'mcp.types.ReadResourceResult'>
Text: Hello, dear client!
Data returned: meta=None contents=[TextResourceContents(uri=AnyUrl('pozdrav://Pavel'), mimeType='text/plain', text='Hello, Pavel!')]
Type: <class 'mcp.types.ReadResourceResult'>
Text: Hello, Pavel!
10!= 3628800
Client has finished in 3.0988974571228027 seconds
Výsledek pro nejpomalejší MCP klient:
Client initialization
Data returned: meta=None contents=[TextResourceContents(uri=AnyUrl('pozdrav://'), mimeType='text/plain', text='Hello, dear client!')]
Type: <class 'mcp.types.ReadResourceResult'>
Text: Hello, dear client!
Data returned: meta=None contents=[TextResourceContents(uri=AnyUrl('pozdrav://Pavel'), mimeType='text/plain', text='Hello, Pavel!')]
Type: <class 'mcp.types.ReadResourceResult'>
Text: Hello, Pavel!
10!= 3628800
Client has finished in 14.61089015007019 seconds
Z tohoto času (cca 1,5×10 sekund) vyplývá, že většina obsluhy klientů MCP serverem je prováděna sekvenčně a nikoli souběžně.
10. Asynchronní souběžný přístup ke „zpomalenému“ MCP serveru
Ovšem souběžný přístup je možné v případě potřeby zajistit i přímo v realizaci MCP klienta. Namísto jediného spuštění asynchronní funkce run jich necháme spustit několik. Pro tento účel si vytvoříme pomocnou asynchronní funkci main, v níž několikrát spustíme run a budeme čekat na všechny výsledky:
async def main():
# spuštění několika klientů
clients = [run(i) for i in range(10)]
await asyncio.gather(*clients)
Upravený zdrojový kód klienta (včetně měření času) bude vypadat následovně:
"""MCP klient, který přečte zvolený zdroj a získá z něj data."""
from time import time
from mcp import ClientSession, StdioServerParameters
from mcp.client.sse import sse_client
async def run(i):
"""Realizace klienta."""
async with sse_client(url="http://localhost:8000/sse") as (read, write):
async with ClientSession(read, write) as session:
t1 = time()
print(f"Client #{i} initialization")
print()
await session.initialize()
# přečtení zdroje bez selektoru
data = await session.read_resource("pozdrav://")
print("Data returned:", data)
print("Type:", type(data))
text = data.contents[0].text
print("Text:", text)
print()
# přečtení zdroje se selektorem
data = await session.read_resource("pozdrav://Pavel")
print("Data returned:", data)
print("Type:", type(data))
text = data.contents[0].text
print("Text:", text)
print()
# zavolání nástroje
data = await session.call_tool("factorial", arguments={"n": 10})
factorial = data.content[0].text
print("10!=", factorial)
print()
t2 = time()
difftime = t2 - t1
print(f"Client has finished in {difftime} seconds")
print()
async def main():
# spuštění několika klientů
clients = [run(i) for i in range(10)]
await asyncio.gather(*clients)
if __name__ == "__main__":
import asyncio
asyncio.run(main())
Opět se podívejme na výsledky, zejména na celkové časy:
Client #0 initialization
Client #1 initialization
Client #2 initialization
Client #7 initialization
Client #8 initialization
Client #9 initialization
Client #6 initialization
Client #4 initialization
Client #3 initialization
Client #5 initialization
Data returned: meta=None contents=[TextResourceContents(uri=AnyUrl('pozdrav://'), mimeType='text/plain', text='Hello, dear client!')]
Type: <class 'mcp.types.ReadResourceResult'>
Text: Hello, dear client!
Data returned: meta=None contents=[TextResourceContents(uri=AnyUrl('pozdrav://'), mimeType='text/plain', text='Hello, dear client!')]
Type: <class 'mcp.types.ReadResourceResult'>
Text: Hello, dear client!
Data returned: meta=None contents=[TextResourceContents(uri=AnyUrl('pozdrav://'), mimeType='text/plain', text='Hello, dear client!')]
Type: <class 'mcp.types.ReadResourceResult'>
Text: Hello, dear client!
Data returned: meta=None contents=[TextResourceContents(uri=AnyUrl('pozdrav://'), mimeType='text/plain', text='Hello, dear client!')]
Type: <class 'mcp.types.ReadResourceResult'>
Text: Hello, dear client!
Data returned: meta=None contents=[TextResourceContents(uri=AnyUrl('pozdrav://'), mimeType='text/plain', text='Hello, dear client!')]
Type: <class 'mcp.types.ReadResourceResult'>
Text: Hello, dear client!
Data returned: meta=None contents=[TextResourceContents(uri=AnyUrl('pozdrav://'), mimeType='text/plain', text='Hello, dear client!')]
Type: <class 'mcp.types.ReadResourceResult'>
Text: Hello, dear client!
Data returned: meta=None contents=[TextResourceContents(uri=AnyUrl('pozdrav://'), mimeType='text/plain', text='Hello, dear client!')]
Type: <class 'mcp.types.ReadResourceResult'>
Text: Hello, dear client!
Data returned: meta=None contents=[TextResourceContents(uri=AnyUrl('pozdrav://'), mimeType='text/plain', text='Hello, dear client!')]
Type: <class 'mcp.types.ReadResourceResult'>
Text: Hello, dear client!
Data returned: meta=None contents=[TextResourceContents(uri=AnyUrl('pozdrav://'), mimeType='text/plain', text='Hello, dear client!')]
Type: <class 'mcp.types.ReadResourceResult'>
Text: Hello, dear client!
Data returned: meta=None contents=[TextResourceContents(uri=AnyUrl('pozdrav://'), mimeType='text/plain', text='Hello, dear client!')]
Type: <class 'mcp.types.ReadResourceResult'>
Text: Hello, dear client!
Data returned: meta=None contents=[TextResourceContents(uri=AnyUrl('pozdrav://Pavel'), mimeType='text/plain', text='Hello, Pavel!')]
Type: <class 'mcp.types.ReadResourceResult'>
Text: Hello, Pavel!
Data returned: meta=None contents=[TextResourceContents(uri=AnyUrl('pozdrav://Pavel'), mimeType='text/plain', text='Hello, Pavel!')]
Type: <class 'mcp.types.ReadResourceResult'>
Text: Hello, Pavel!
Data returned: meta=None contents=[TextResourceContents(uri=AnyUrl('pozdrav://Pavel'), mimeType='text/plain', text='Hello, Pavel!')]
Type: <class 'mcp.types.ReadResourceResult'>
Text: Hello, Pavel!
Data returned: meta=None contents=[TextResourceContents(uri=AnyUrl('pozdrav://Pavel'), mimeType='text/plain', text='Hello, Pavel!')]
Type: <class 'mcp.types.ReadResourceResult'>
Text: Hello, Pavel!
Data returned: meta=None contents=[TextResourceContents(uri=AnyUrl('pozdrav://Pavel'), mimeType='text/plain', text='Hello, Pavel!')]
Type: <class 'mcp.types.ReadResourceResult'>
Text: Hello, Pavel!
Data returned: meta=None contents=[TextResourceContents(uri=AnyUrl('pozdrav://Pavel'), mimeType='text/plain', text='Hello, Pavel!')]
Type: <class 'mcp.types.ReadResourceResult'>
Text: Hello, Pavel!
Data returned: meta=None contents=[TextResourceContents(uri=AnyUrl('pozdrav://Pavel'), mimeType='text/plain', text='Hello, Pavel!')]
Type: <class 'mcp.types.ReadResourceResult'>
Text: Hello, Pavel!
Data returned: meta=None contents=[TextResourceContents(uri=AnyUrl('pozdrav://Pavel'), mimeType='text/plain', text='Hello, Pavel!')]
Type: <class 'mcp.types.ReadResourceResult'>
Text: Hello, Pavel!
Data returned: meta=None contents=[TextResourceContents(uri=AnyUrl('pozdrav://Pavel'), mimeType='text/plain', text='Hello, Pavel!')]
Type: <class 'mcp.types.ReadResourceResult'>
Text: Hello, Pavel!
Data returned: meta=None contents=[TextResourceContents(uri=AnyUrl('pozdrav://Pavel'), mimeType='text/plain', text='Hello, Pavel!')]
Type: <class 'mcp.types.ReadResourceResult'>
Text: Hello, Pavel!
10!= 3628800
Client has finished in 10.609342098236084 seconds
10!= 3628800
Client has finished in 15.126225709915161 seconds
10!= 3628800
Client has finished in 15.123616933822632 seconds
10!= 3628800
Client has finished in 15.123651266098022 seconds
10!= 3628800
Client has finished in 15.126556634902954 seconds
10!= 3628800
Client has finished in 15.12637448310852 seconds
10!= 3628800
Client has finished in 15.12638545036316 seconds
10!= 3628800
Client has finished in 15.126460552215576 seconds
10!= 3628800
Client has finished in 15.133534669876099 seconds
10!= 3628800
Client has finished in 15.127150774002075 seconds
I zde je patrné, že MCP server nedokáže obsloužit více požadavků současně.
11. Podpora rastrových obrázků v balíčku mcp
Knihovna mcp do jisté míry podporuje i posílání rastrových obrázků z MCP serveru klientovi. To je obecně užitečná vlastnost, protože mnoho generativních AI vytváří či modifikuje nějaké formy rastrových dat. K poslání rastrového obrázku MCP serverem se využívá knihovna Pillow resp. PIL (druhé jméno se používá z důvodů dodržení kompatibility). Má to však dvě omezení:
- Obrázky prozatím nelze použít ve funkci zdrojů (resource), což by však mělo být doimplementováno
- Obrázky je nutné na straně klienta dekódovat, což si pochopitelně ukážeme
12. MCP server posílající rastrový obrázek
Podívejme se nyní, jak bude vypadat realizace nástroje (tool), který jako výsledek své činnosti vytvoří a vrátí rastrový obrázek. Hlavička funkce s realizací takového nástroje vypadá následovně:
@mcp.tool() def house() -> Image:
přičemž typ Image je importován z balíčku mcp.server.fastmcp.
Ve funkci zkonstruujeme nový binární obrázek o rozlišení 512×512 pixelů:
image = PILImage.new("1", (512, 512))
A vykreslíme do rastrového obrázku několik úseček – klasický domeček „jedním tahem“:
draw = ImageDraw.Draw(image) ... ... ... draw.line(endpoints, fill=255)
Nyní nám již zbývá data rastrového obrázku vrátit klientovi. To je snadné:
return Image(data=image.tobytes(), format="png")
Úplná implementace MCP serveru s nástrojem vracejícím obrázek vypadá takto:
"""MCP server se nástrojem s obrázkem."""
from PIL import Image as PILImage
from PIL import ImageDraw
from mcp.server.fastmcp import FastMCP, Image
# konstrukce serveru
mcp = FastMCP("Test")
#@mcp.resource("house://", mime_type="image/png")
@mcp.tool()
def house() -> Image:
"""Vrací rastrový obrázek."""
# vytvoření prázdného obrázku
image = PILImage.new("1", (512, 512))
# objekt umožňující kreslení do obrázku
draw = ImageDraw.Draw(image)
endpoints = [
100,
500,
400,
200,
100,
200,
250,
50,
400,
200,
400,
500,
100,
200,
100,
500,
400,
500,
]
draw.line(endpoints, fill=255)
return Image(data=image.tobytes(), format="png")
# přímé spuštění serveru v režimu SSE (Server-Sent Events)
if __name__ == "__main__":
mcp.run(transport="sse")
13. Dekódování obrázku na straně MCP klienta
Realizace MCP klienta, který přes MCP server zavolá nástroj a získá příslušný obrázek, vyžaduje explicitní programovou konverzi, kterou si ukážeme. Nejdříve pochopitelně zavoláme nástroj a přečteme data, která nástroj vrátil:
data = await session.call_tool("house")
Data jsou zakódována do BASE64 a poslána jako běžný text (to je ovšem zvláštní, protože existuje i možnost vrácení binárních dat). Proto provedeme získání první textové odpovědi, dekódování tohoto textu z BASE64 do binárních dat a rekonstrukce původního obrázku:
image = PILImage.frombytes('1', (512,512), base64.b64decode(data.content[0].data))
Nyní je již možné nechat si obrázek vykreslit na obrazovce, uložit do souboru atd.:
image.save("house.png")
Úplný zdrojový kód MCP klienta tedy může vypadat takto:
"""MCP klient, který přečte obrázek."""
import base64
from PIL import Image as PILImage
from mcp import ClientSession
from mcp.client.sse import sse_client
async def run():
"""Realizace klienta."""
async with sse_client(url="http://localhost:8000/sse") as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
data = await session.call_tool("house")
image = PILImage.frombytes('1', (512,512), base64.b64decode(data.content[0].data))
print(image)
image.save("house.png")
# přímé spuštění klienta
if __name__ == "__main__":
import asyncio
asyncio.run(run())
Po inicializaci klienta by se měla vypsat „hlavička“ objektu reprezentujícího přečtený rastrový obrázek:
<PIL.Image.Image image mode=1 size=512x512 at 0x7F9589688E30>
Rastrový obrázek, který z MCP serveru získáme a uložíme do souboru ve formátu PNG, by měl vypadat takto:
Obrázek 1: Rastrový obrázek vráceny MCP serverem a uložený do souboru ve formátu PNG.
13. Další možnosti nabízené protokolem MCP
14. Repositář s demonstračními příklady
Zdrojové kódy všech prozatím popsaných demonstračních příkladů určených pro Python a balíček mcp byly uloženy do repositáře dostupného na adrese https://github.com/tisnik/most-popular-python-libs. V tabulce zobrazené níže jsou odkazy na jednotlivé příklady:
15. Příloha: články o knihovně PIL/Pillow
Dnes jsme se na okraj zmínili o knihovně PIL resp. Pillow, která v Pythonu slouží pro manipulaci s rastrovými obrázky, a to včetně načítání a ukládání těchto obrázků do souborů ve vybraných formátech (PNG, GIF, JPEG, PBM atd.), ale i pro kreslení jednoduchých 2D tvarů. Možnostmi poskytovanými touto knihovnou jsem se zabýval v následující dvojici článků:
- Užitečné knihovny a moduly pro Python: kreslení a pokročilé manipulace s obrázky v knihovně Pillow
https://mojefedora.cz/uzitecne-knihovny-a-moduly-pro-python-kresleni-a-pokrocile-manipulace-s-obrazky-v-knihovne-pillow/ - Užitečné knihovny a moduly pro Python: knihovna Pillow určená pro manipulaci s rastrovými obrázky
https://mojefedora.cz/uzitecne-knihovny-a-moduly-pro-python-knihovna-pillow-urcena-pro-manipulaci-s-rastrovymi-obrazky/
16. Odkazy na Internetu
- MCP Python SDK
https://github.com/modelcontextprotocol/python-sdk?tab=readme-ov-file#running-your-server - MCP protocol: Resources
https://modelcontextprotocol.info/docs/concepts/resources/ - Example Servers
https://modelcontextprotocol.io/examples - Core architecture
https://modelcontextprotocol.io/docs/concepts/architecture - Unleashing the Power of Model Context Protocol (MCP): A Game-Changer in AI Integration
https://techcommunity.microsoft.com/blog/educatordeveloperblog/unleashing-the-power-of-model-context-protocol-mcp-a-game-changer-in-ai-integrat/4397564 - MPC inspector
https://github.com/modelcontextprotocol/inspector - Model Context Protocol servers
https://github.com/modelcontextprotocol/servers - python-sdk na GitHubu
https://github.com/modelcontextprotocol/python-sdk - typescript-sdk na GitHubu
https://github.com/modelcontextprotocol/typescript-sdk - mcp-golang
https://github.com/metoro-io/mcp-golang - MCP server: A step-by-step guide to building from scratch
https://composio.dev/blog/mcp-server-step-by-step-guide-to-building-from-scrtch/ - How to Build an MCP Server Fast: A Step-by-Step Tutorial
https://medium.com/@eugenesh4work/how-to-build-an-mcp-server-fast-a-step-by-step-tutorial-e09faa5f7e3b - Step-by-Step Guide: Building an MCP Server using Python-SDK, AlphaVantage & Claude AI
https://medium.com/@syed_hasan/step-by-step-guide-building-an-mcp-server-using-python-sdk-alphavantage-claude-ai-7a2bfb0c3096 - RFC 6570: URI Template
https://datatracker.ietf.org/doc/html/rfc6570 - Return resources as structured JSON instead of text?
https://github.com/modelcontextprotocol/python-sdk/issues/279 - Python standard library: pprint
https://docs.python.org/3/library/pprint.html - Python standard library: json — JSON encoder and decoder¶
https://docs.python.org/3/library/json.html - Calling MCP Servers the Hard Way
https://deadprogrammersociety.com/2025/03/calling-mcp-servers-the-hard-way.html - mcptools
https://github.com/f/mcptools - Server-sent events
https://en.wikipedia.org/wiki/Server-sent_events - Model context protocol (MCP)
https://openai.github.io/openai-agents-python/mcp/ - A Clear Intro to MCP (Model Context Protocol) with Code Examples
https://towardsdatascience.com/clear-intro-to-mcp/ - A Developer's Guide to the MCP
https://www.getzep.com/ai-agents/developer-guide-to-mcp - MCP: Flash in the Pan or Future Standard?
https://blog.langchain.dev/mcp-fad-or-fixture/ - MCP yeah you know me: A complete guide and review of Model Context Protocol (MCP)
https://ebi.ai/blog/model-context-protocol-guide/ - Pillow documentation
https://pillow.readthedocs.io/en/stable/handbook/tutorial.html - Pillow: Python Imaging Library (Fork)
https://pypi.org/project/pillow/ - How to use Pillow, a fork of PIL
https://www.pythonforbeginners.com/gui/how-to-use-pillow - PNG is Not GIF
https://www.root.cz/clanky/png-is-not-gif/
ČLÁNKY DO MAILU