Run the OAuth client example.
()
| 57 | |
| 58 | |
| 59 | async def main(): |
| 60 | """Run the OAuth client example.""" |
| 61 | oauth_auth = OAuthClientProvider( |
| 62 | server_url="http://localhost:8001", |
| 63 | client_metadata=OAuthClientMetadata( |
| 64 | client_name="Example MCP Client", |
| 65 | redirect_uris=[AnyUrl("http://localhost:3000/callback")], |
| 66 | grant_types=["authorization_code", "refresh_token"], |
| 67 | response_types=["code"], |
| 68 | scope="user", |
| 69 | ), |
| 70 | storage=InMemoryTokenStorage(), |
| 71 | redirect_handler=handle_redirect, |
| 72 | callback_handler=handle_callback, |
| 73 | ) |
| 74 | |
| 75 | async with httpx.AsyncClient(auth=oauth_auth, follow_redirects=True) as custom_client: |
| 76 | async with streamable_http_client("http://localhost:8001/mcp", http_client=custom_client) as (read, write): |
| 77 | async with ClientSession(read, write) as session: |
| 78 | await session.initialize() |
| 79 | |
| 80 | tools = await session.list_tools() |
| 81 | print(f"Available tools: {[tool.name for tool in tools.tools]}") |
| 82 | |
| 83 | resources = await session.list_resources() |
| 84 | print(f"Available resources: {[r.uri for r in resources.resources]}") |
| 85 | |
| 86 | |
| 87 | def run(): |
no test coverage detected