Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion packages/fastmcp/examples/delegated_access/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -163,7 +163,7 @@ The example demonstrates comprehensive error handling patterns:
| `has_errors()` | Check for any errors (global or resource-specific) |
| `get_errors()` | Get all error details as a dictionary |
| `has_resource_error(url)` | Check for errors on a specific resource |
| `get_resource_errors(url)` | Get errors for a specific resource |
| `get_resource_error(url)` | Get errors for a specific resource |
| `has_error()` | Check for global errors only |
| `get_error()` | Get global error details |

Expand Down
4 changes: 2 additions & 2 deletions packages/fastmcp/examples/delegated_access/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -100,7 +100,7 @@ async def list_github_repos(ctx: Context, per_page: int = 5) -> dict:

Demonstrates:
- Resource-specific error checking with has_resource_error()
- Getting resource-specific errors with get_resource_errors()
- Getting resource-specific errors with get_resource_error()
- Parameterized API calls

Args:
Expand All @@ -114,7 +114,7 @@ async def list_github_repos(ctx: Context, per_page: int = 5) -> dict:

# Check for resource-specific error (alternative to has_errors())
if access_context.has_resource_error("https://api.github.com"):
resource_errors = access_context.get_resource_errors("https://api.github.com")
resource_errors = access_context.get_resource_error("https://api.github.com")
return {
"message": "Token exchange failed for GitHub API",
"details": resource_errors,
Expand Down
4 changes: 2 additions & 2 deletions packages/fastmcp/src/keycardai/fastmcp/provider.py
Original file line number Diff line number Diff line change
Expand Up @@ -233,7 +233,7 @@ def get_error(self) -> dict[str, str] | None:
"""Get global error if any."""
return self._error

def get_resource_errors(self, resource: str) -> dict[str, str] | None:
def get_resource_error(self, resource: str) -> dict[str, str] | None:
"""Get error for a specific resource."""
return self._resource_errors.get(resource)

Expand Down Expand Up @@ -279,7 +279,7 @@ def access(self, resource: str) -> TokenResponse:
raise ResourceAccessError(
resource=resource,
error_type="resource_error",
error_details=self.get_resource_errors(resource)
error_details=self.get_resource_error(resource)
)

# Check if token exists
Expand Down
4 changes: 2 additions & 2 deletions packages/fastmcp/tests/integration/test_grant_decorator.py
Original file line number Diff line number Diff line change
Expand Up @@ -129,7 +129,7 @@ async def test_function(ctx: Context, user_id: str):
# Check if there's a resource error
access_ctx = await ctx.get_state("keycardai")
if access_ctx.has_resource_error("https://api.example.com"):
error = access_ctx.get_resource_errors("https://api.example.com")
error = access_ctx.get_resource_error("https://api.example.com")
return {"error": error["message"], "isError": True}
return {"error": "No error", "isError": False, "access_ctx": access_ctx}

Expand Down Expand Up @@ -292,7 +292,7 @@ def test_access_context_error_states(self):
})
assert access_context.has_resource_error("https://api1.com")
assert access_context.get_status() == "partial_error"
assert access_context.get_resource_errors("https://api1.com")["message"] == "Resource failed"
assert access_context.get_resource_error("https://api1.com")["message"] == "Resource failed"

def test_access_context_partial_success(self):
"""Test AccessContext with partial success scenario."""
Expand Down
84 changes: 84 additions & 0 deletions packages/fastmcp/tests/test_access_context.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,84 @@
"""Unit tests for the fastmcp provider AccessContext.

The fastmcp AccessContext.access() is an independent implementation in
provider.py with the same three error paths as the oauth AccessContext.
These tests mirror packages/oauth/tests/.../test_access_context.py and
verify the rich-error contract holds for the fastmcp provider too.
"""

import pytest

from keycardai.fastmcp.provider import AccessContext
from keycardai.mcp.server.exceptions import ResourceAccessError
from keycardai.oauth.types.models import TokenResponse


def _token() -> TokenResponse:
return TokenResponse(access_token="tok", token_type="bearer")


def test_access_returns_token_when_present():
ctx = AccessContext()
ctx.set_token("https://api.example.com", _token())
assert ctx.access("https://api.example.com").access_token == "tok"


def test_access_carries_missing_token_context():
ctx = AccessContext()
ctx.set_token("https://api.example.com", _token())

with pytest.raises(ResourceAccessError) as exc_info:
ctx.access("https://missing.example.com")

err = exc_info.value
assert err.details["error_type"] == "missing_token"
assert err.details["requested_resource"] == "https://missing.example.com"
assert set(err.details["available_resources"]) == {"https://api.example.com"}


def test_access_carries_resource_error_context():
ctx = AccessContext()
detail = {"message": "denied by AS", "code": "access_denied"}
ctx.set_resource_error("https://api.example.com", detail)

with pytest.raises(ResourceAccessError) as exc_info:
ctx.access("https://api.example.com")

err = exc_info.value
assert err.details["error_type"] == "resource_error"
assert err.details["requested_resource"] == "https://api.example.com"
assert err.details["error_details"] == detail
assert err.details["available_resources"] is None


def test_access_carries_global_error_context():
ctx = AccessContext()
detail = {"message": "token exchange failed"}
ctx.set_error(detail)

with pytest.raises(ResourceAccessError) as exc_info:
ctx.access("https://api.example.com")

err = exc_info.value
assert err.details["error_type"] == "global_error"
assert err.details["requested_resource"] == "https://api.example.com"
assert err.details["error_details"] == detail
assert err.details["available_resources"] is None


def test_get_resource_error_returns_stored_detail_or_none():
ctx = AccessContext()
ctx.set_resource_error("https://api.example.com", {"message": "transient"})
assert ctx.get_resource_error("https://api.example.com") == {"message": "transient"}
assert ctx.get_resource_error("https://other.example.com") is None


def test_status_transitions():
ctx = AccessContext()
assert ctx.get_status() == "success"

ctx.set_resource_error("https://api.example.com", {"message": "boom"})
assert ctx.get_status() == "partial_error"

ctx.set_error({"message": "global boom"})
assert ctx.get_status() == "error"
2 changes: 1 addition & 1 deletion packages/mcp-fastmcp/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -222,7 +222,7 @@ if access_context.has_error():

# Check for specific resource errors
if access_context.has_resource_error("https://api.example.com"):
resource_error = access_context.get_resource_errors("https://api.example.com")
resource_error = access_context.get_resource_error("https://api.example.com")

# Get all errors (global + resource-specific)
all_errors = access_context.get_errors()
Expand Down
2 changes: 1 addition & 1 deletion packages/mcp/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -620,7 +620,7 @@ def multi_resource_tool(access_ctx: AccessContext, ctx: Context):

# Handle failed resources
for resource in access_ctx.get_failed_resources():
error = access_ctx.get_resource_errors(resource)
error = access_ctx.get_resource_error(resource)
results[resource] = {"error": error["message"]}

return results
Expand Down
2 changes: 1 addition & 1 deletion packages/mcp/examples/delegated_access/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -193,7 +193,7 @@ The example demonstrates comprehensive error handling patterns:
| `has_errors()` | Check for any errors (global or resource-specific) |
| `get_errors()` | Get all error details as a dictionary |
| `has_resource_error(url)` | Check for errors on a specific resource |
| `get_resource_errors(url)` | Get errors for a specific resource |
| `get_resource_error(url)` | Get errors for a specific resource |
| `has_error()` | Check for global errors only |
| `get_error()` | Get global error details |

Expand Down
4 changes: 2 additions & 2 deletions packages/mcp/examples/delegated_access/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -102,7 +102,7 @@ async def list_github_repos(access_ctx: AccessContext, ctx: Context, per_page: i

Demonstrates:
- Resource-specific error checking with has_resource_error()
- Getting resource-specific errors with get_resource_errors()
- Getting resource-specific errors with get_resource_error()
- Parameterized API calls with AccessContext as first parameter

Args:
Expand All @@ -115,7 +115,7 @@ async def list_github_repos(access_ctx: AccessContext, ctx: Context, per_page: i
"""
# Check for resource-specific error (alternative to has_errors())
if access_ctx.has_resource_error("https://api.github.com"):
resource_errors = access_ctx.get_resource_errors("https://api.github.com")
resource_errors = access_ctx.get_resource_error("https://api.github.com")
return {
"message": "Token exchange failed for GitHub API",
"details": resource_errors,
Expand Down
4 changes: 2 additions & 2 deletions packages/mcp/tests/integration/test_grant_decorator.py
Original file line number Diff line number Diff line change
Expand Up @@ -139,7 +139,7 @@ async def test_grant_decorator_token_exchange_failure_with_injected_client(self,
def test_function(access_ctx: AccessContext, ctx: Context, user_id: str):
# Check if there's a resource error
if access_ctx.has_resource_error("https://api.example.com"):
error = access_ctx.get_resource_errors("https://api.example.com")
error = access_ctx.get_resource_error("https://api.example.com")
return {"error": error["message"], "isError": True}
return {"error": "No error", "isError": False, "access_ctx": access_ctx}

Expand Down Expand Up @@ -287,7 +287,7 @@ def test_access_context_error_states(self):
})
assert access_context.has_resource_error("https://api1.com")
assert access_context.get_status() == "partial_error"
assert access_context.get_resource_errors("https://api1.com")["message"] == "Resource failed"
assert access_context.get_resource_error("https://api1.com")["message"] == "Resource failed"

def test_access_context_partial_success(self):
"""Test AccessContext with partial success scenario."""
Expand Down
20 changes: 16 additions & 4 deletions packages/oauth/src/keycardai/oauth/server/access_context.py
Original file line number Diff line number Diff line change
Expand Up @@ -62,7 +62,7 @@ def get_error(self) -> dict[str, str] | None:
"""Get global error if any."""
return self._error

def get_resource_errors(self, resource: str) -> dict[str, str] | None:
def get_resource_error(self, resource: str) -> dict[str, str] | None:
"""Get error for a specific resource."""
return self._resource_errors.get(resource)

Expand Down Expand Up @@ -96,12 +96,24 @@ def access(self, resource: str) -> TokenResponse:
ResourceAccessError: If resource was not granted or has an error
"""
if self.has_error():
raise ResourceAccessError()
raise ResourceAccessError(
resource=resource,
error_type="global_error",
error_details=self.get_error(),
)

if self.has_resource_error(resource):
raise ResourceAccessError()
raise ResourceAccessError(
resource=resource,
error_type="resource_error",
error_details=self.get_resource_error(resource),
)

if resource not in self._access_tokens:
raise ResourceAccessError()
raise ResourceAccessError(
resource=resource,
error_type="missing_token",
available_resources=list(self._access_tokens.keys()),
)

return self._access_tokens[resource]
2 changes: 1 addition & 1 deletion packages/oauth/src/keycardai/oauth/server/exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -307,7 +307,7 @@ def __init__(
details = {
"requested_resource": resource or "unknown",
"error_type": error_type or "unknown",
"available_resources": available_resources or [],
"available_resources": available_resources if error_type == "missing_token" else None,
"error_details": error_details or {},
"solution": (
"Fix authentication issues before accessing resources"
Expand Down
82 changes: 82 additions & 0 deletions packages/oauth/tests/keycardai/oauth/server/test_access_context.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,82 @@
"""Unit tests for AccessContext.

Covers the rich-error context surfaced through `access()` and the
get_resource_error getter.
"""

import pytest

from keycardai.oauth.server.access_context import AccessContext
from keycardai.oauth.server.exceptions import ResourceAccessError
from keycardai.oauth.types.models import TokenResponse


def _token() -> TokenResponse:
return TokenResponse(access_token="tok", token_type="bearer")


def test_access_returns_token_when_present():
ctx = AccessContext()
ctx.set_token("https://api.example.com", _token())
assert ctx.access("https://api.example.com").access_token == "tok"


def test_access_carries_missing_token_context():
ctx = AccessContext()
ctx.set_token("https://api.example.com", _token())

with pytest.raises(ResourceAccessError) as exc_info:
ctx.access("https://missing.example.com")

err = exc_info.value
assert err.details["error_type"] == "missing_token"
assert err.details["requested_resource"] == "https://missing.example.com"
assert set(err.details["available_resources"]) == {"https://api.example.com"}


def test_access_carries_resource_error_context():
ctx = AccessContext()
detail = {"message": "denied by AS", "code": "access_denied"}
ctx.set_resource_error("https://api.example.com", detail)

with pytest.raises(ResourceAccessError) as exc_info:
ctx.access("https://api.example.com")

err = exc_info.value
assert err.details["error_type"] == "resource_error"
assert err.details["requested_resource"] == "https://api.example.com"
assert err.details["error_details"] == detail
assert err.details["available_resources"] is None


def test_access_carries_global_error_context():
ctx = AccessContext()
detail = {"message": "token exchange failed"}
ctx.set_error(detail)

with pytest.raises(ResourceAccessError) as exc_info:
ctx.access("https://api.example.com")

err = exc_info.value
assert err.details["error_type"] == "global_error"
assert err.details["requested_resource"] == "https://api.example.com"
assert err.details["error_details"] == detail
assert err.details["available_resources"] is None


def test_get_resource_error_returns_stored_detail_or_none():
ctx = AccessContext()
ctx.set_resource_error("https://api.example.com", {"message": "transient"})
assert ctx.get_resource_error("https://api.example.com") == {"message": "transient"}
assert ctx.get_resource_error("https://other.example.com") is None


def test_status_transitions():
ctx = AccessContext()
assert ctx.get_status() == "success"

ctx.set_resource_error("https://api.example.com", {"message": "boom"})
assert ctx.get_status() == "partial_error"

ctx.set_error({"message": "global boom"})
assert ctx.get_status() == "error"
Loading